我使用的是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'
1条答案
按热度按时间prdp8dxp1#
? 一只可怕的虫子?
虫子
我认为如果你在进程执行lpop时终止它,就会发生这种情况
SEMAPHORE:test:AVAILABLE
很可能在这里https://github.com/dv/redis-semaphore/blob/v0.3.1/lib/redis/semaphore.rb#l67去复制它
现在,您首先有:
在上述代码之后,您将得到:
代码检查
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)
end
end
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
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"}