redis信号量锁不能被释放

rpppsulh  于 2021-06-09  发布在  Redis
关注(0)|答案(1)|浏览(421)

我使用的是redis信号量gem版本 0.3.1 .
由于某些原因,我偶尔不能释放一个过时的redis锁。根据我的分析,如果我的docker进程在锁创建后崩溃,那么这种情况似乎会发生。
我已经在下面描述了我的调试过程,想知道是否有人可以建议如何进一步调试。
假设我们要创建一个具有此名称的redis锁:

name = "test"

我们在两个不同的终端窗口中插入这个变量。首先,我们运行:

def lock_for_15_secs(name)
  job = Redis::Semaphore.new(name.to_sym, redis: NonBlockingRedis.new(), custom_blpop: true, :stale_client_timeout => 15)
  if job.lock(-1) == "0"
    puts "Locked and starting"
    sleep(15)
    puts "Now it's stale, try to release in another process"
    sleep(15)
    puts "Now trying to unlock"
    unlock = job.unlock
    puts unlock == false ? "Wuhuu, already unlocked" : "Hm, should have been unlocked by another process, but wasn't"
  end
end
lock_for_15_secs(name)

在第二次跑步中:

def release_and_lock(name)
  job     = Redis::Semaphore.new(name.to_sym, redis: NonBlockingRedis.new(), custom_blpop: true, :stale_client_timeout => 15)
  release = job.release_stale_locks!
  count   = job.available_count
  puts "Release reponse is #{release.inspect} and available count is #{count}"
  if job.lock(-1) == "0"
    puts "Wuhuu, we can lock it"
    job.unlock
  else
    puts "Hmm, we can't lock it"
  end
end
release_and_lock(name)

这通常按预期进行。在15秒钟内,第二个终端无法释放锁,但在15秒钟后运行时,它会释放。下面是 release_and_lock(name) .
15秒前:

irb(main):1:0> release_and_lock(name)
Release reponse is {"0"=>"1580292557.321834"} and available count is 0
Hmm, we can't lock it
=> nil

15秒后:

irb(main):2:0> release_and_lock(name)
Release reponse is {"0"=>"1580292557.321834"} and available count is 1
Wuhuu, we can lock it
=> 1
irb(main):3:0> release_and_lock(name)
Release reponse is {} and available count is 1
Wuhuu, we can lock it

但是每当我看到一个过时的锁没有被释放,我就跑 release_and_lock(name) 要进行诊断,将返回:

irb(main):4:0> release_and_lock(name)
Release reponse is {} and available count is 0
Hmm, we can't lock it

在这一点上,我唯一的选择就是刷新redis:

require 'non_blocking_redis'
non_blocking_redis = NonBlockingRedis.new()
non_blocking_redis.flushall

p、 s.我的 NonBlockingRedis 继承自 Redis :

class Redis
  class Semaphore
    def initialize(name, opts = {})
      @custom_opts          = opts
      @name                 = name
      @resource_count       = opts.delete(:resources) || 1
      @stale_client_timeout = opts.delete(:stale_client_timeout)
      @redis                = opts.delete(:redis) || Redis.new(opts)
      @use_local_time       = opts.delete(:use_local_time)
      @custom_blpop         = opts.delete(:custom_blpop) # false=queue, true=cancel
      @tokens               = []
    end

    def lock(timeout = 0)
      exists_or_create!
      release_stale_locks! if check_staleness?
      token_pair    = @redis.blpop(available_key, timeout, @custom_blpop)
      return false if token_pair.nil?
      current_token = token_pair[1]
      @tokens.push(current_token)
      @redis.hset(grabbed_key, current_token, current_time.to_f)
      if block_given?
        begin
          yield current_token
        ensure
          signal(current_token)
        end
      end
      current_token
    end
    alias_method :wait, :lock
  end
end

class NonBlockingRedis < Redis

  def initialize(options = {})
    if options.empty?
      options = {
        url: Rails.application.secrets.redis_url,
        db:  Rails.application.secrets.redis_sidekiq_db,
        driver: :hiredis,
        network_timeout: 5
      }
    end

    super(options)
  end

  def blpop(key, timeout, custom_blpop)
    if custom_blpop
      if timeout == -1
        result = lpop(key)
        return result if result.nil?
        return [key, result]
      else
        super(key, timeout)
      end
    else
       super
    end
  end

  def lock(timeout = 0)
    exists_or_create!
    release_stale_locks! if check_staleness?
    token_pair = @redis.blpop(available_key, timeout, @custom_blpop)
    return false if token_pair.nil?
    current_token = token_pair[1]
    @tokens.push(current_token)
    @redis.hset(grabbed_key, current_token, current_time.to_f)
    if block_given?
      begin
        yield current_token
      ensure
        signal(current_token)
      end
    end
    current_token
  end
  alias_method :wait, :lock
end

require 'non_blocking_redis'
prdp8dxp

prdp8dxp1#

? 一只可怕的虫子?

虫子

我认为如果你在进程执行lpop时终止它,就会发生这种情况 SEMAPHORE:test:AVAILABLE 很可能在这里https://github.com/dv/redis-semaphore/blob/v0.3.1/lib/redis/semaphore.rb#l67
去复制它

NonBlockingRedis.new.flushall

release_and_lock('test');

NonBlockingRedis.new.lpop('SEMAPHORE:test:AVAILABLE')

现在,您首先有:

SEMAPHORE:test:AVAILABLE                0
SEMAPHORE:test:VERSION          1
SEMAPHORE:test:EXISTS           1

在上述代码之后,您将得到:

SEMAPHORE:test:VERSION          1
SEMAPHORE:test:EXISTS           1

代码检查 SEMAPHORE:test:EXISTS 然后期望 SEMAPHORE:test:AVAILABLE / SEMAPHORE:test:GRABBED ##解决方案
从我的简短检查,我不认为有可能使宝石工作没有修改。我试着添加一个 expiration: 但不知怎的,它成功地禁用了 SEMAPHORE:test:EXISTS ```
NonBlockingRedis.new.ttl('SEMAPHORE:test:EXISTS') # => -1 and it should have been e.g. 20 seconds and going down

所以。。也许会有个解决办法

class Redis
class Semaphore
def exists_or_create!
token = @redis.getset(exists_key, EXISTS_TOKEN)

  if token.nil? || all_tokens.empty?
    create!
  else
    # Previous versions of redis-semaphore did not set `version_key`.
    # Make sure it's set now, so we can use it in future versions.

    if token == API_VERSION && @redis.get(version_key).nil?
      @redis.set(version_key, API_VERSION)
    end

    true
  end
end

end
end

所有的代币都是https://github.com/dv/redis-semaphore/blob/v0.3.1/lib/redis/semaphore.rb#l120
我很快就会打开宝石的公关->https://github.com/dv/redis-semaphore/pull/66 也许 吧?‍♂️

## 附注1

不知道你怎么用这个 `NonBlockingRedis` 但它不适用于 `Redis::Semaphore` . 你知道吗 `lock(-1)` 在代码里是什么意思 `lpop` . 而且代码从不调用 `lock` .

## 随机的

这里有一个帮手来倒钥匙

class Test
def self.all
r = NonBlockingRedis.new
puts r.keys('*').map { |k|
[
k,
((r.hgetall(k) rescue r.get(k)) rescue r.lrange(k, 0, -1).join(' | '))
].join("\t\t")
}
end
end

Test.all

SEMAPHORE:test:AVAILABLE 0
SEMAPHORE:test:VERSION 1
SEMAPHORE:test:EXISTS 1

为了完整起见,这里是当你抓住锁时的样子

SEMAPHORE:test:VERSION 1
SEMAPHORE:test:EXISTS 1
SEMAPHORE:test:GRABBED {"0"=>"1583672948.7168388"}

相关问题