上一篇讲到,RedissonFairLock 它主要是基于 RedissonLock 做的扩展,主要扩展在于加锁和释放锁的地方。所以我们要研究 RedissonFairLock 释放锁的逻辑,只需要关注它重写的 RedissonLock#unlockAsync 方法即可。至于如何计算 slot 来执行 lua 脚本,释放后需停止 wathchdog 的运行,和 RedissonLock 保持一致,我们只要研究了 RedissonLock 释放锁的逻辑后,这里就不再需要关注了。那我们下面直接上lua脚本的分析。
RedissonFairLock#unlockInnerAsync:
@Override protected RFuture<Boolean> unlockInnerAsync(long threadId) { return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, // remove stale threads "while true do " + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end; " + "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" + "if timeout <= tonumber(ARGV[4]) then " + "redis.call('zrem', KEYS[3], firstThreadId2); " + "redis.call('lpop', KEYS[2]); " + "else " + "break;" + "end; " + "end;" + "if (redis.call('exists', KEYS[1]) == 0) then " + "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + "if nextThreadId ~= false then " + "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " + "end; " + "return 1; " + "end;" + "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "end; " + "redis.call('del', KEYS[1]); " + "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + "if nextThreadId ~= false then " + "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " + "end; " + "return 1; ", Arrays.asList(getName(), threadsQueueName, timeoutSetName, getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis()); }
因为基于 RedissonLock 加入了公平获取锁的机制,所以lua脚本还是挺长的,下面我们一步一步分析。
Arrays.asList(getName(), threadsQueueName, timeoutSetName, getChannelName()):
KEYS:["myLock","redisson_lock_queue:{myLock}","redisson_lock_timeout:{myLock}","redisson_lock__channel:{myLock}"]
LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis():
ARGVS:[0L,3w毫秒,"UUID:threadId",当前时间时间戳]
场景:
和获取锁的第一步一样,开个死循环清理过期的等待线程,主要避免下面场景,避免无效客户端占用等待队列资源
lua脚本:
"while true do " + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" + "if firstThreadId2 == false then " + "break;" + "end;" + "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" + "if timeout <= tonumber(ARGV[4]) then " + // remove the item from the queue and timeout set // NOTE we do not alter any other timeout "redis.call('zrem', KEYS[3], firstThreadId2);" + "redis.call('lpop', KEYS[2]);" + "else " + "break;" + "end;" + "end;" +
分析:
开启死循环
利用 lindex 命令判断等待队列中第一个元素是否存在,如果存在,直接跳出循环
lidex redisson_lock_queue:{myLock} 0
如果等待队列中第一个元素不为空(例如返回了LockName,即客户端UUID拼接线程ID),利用 zscore 在 超时记录集合(sorted set) 中获取对应的超时时间
zscore redisson_lock_timeout:{myLock} UUID:threadId
如果超时时间已经小于当前时间,那么首先从超时集合中移除该节点,接着也在等待队列中弹出第一个节点
zrem redisson_lock_timeout:{myLock} UUID:threadId lpop redisson_lock_queue:{myLock}
如果等待队列中的第一个元素还未超时,直接退出死循环
场景:
lua脚本:
"if (redis.call('exists', KEYS[1]) == 0) then " + "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + "if nextThreadId ~= false then " + "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " + "end; " + "return 1; " + "end;"
分析:
利用 exists 命令判断锁是否存在
exists myLock
如果锁不存在,利用 lidex 获取等待队列中的第一个元素
lindex redisson_lock_queue:{myLock} 0
如果等待列表中第一个元素不为空,即还存在等待线程,往等待线程的订阅channel发送消息,通知其可以尝试获取锁了
publish redisson_lock__channel:{myLock}:UUID:threadId 0
关于这个订阅通道,之前介绍 RedissonLock 死循环等待锁的时候没有详细介绍,这个可以补充一下。
RedissonLock 所有等待线程都是订阅锁的同一个channel:redisson_lock__channel:{myLock},当有线程释放锁的时候,会往这个通道发送消息,此时所有等待现成都可以订阅消费这条消息,从而从等待状态中释放出来,重新尝试获取锁。
而 RedissonFairLock 不太一样,因为它要支持公平获取锁,即先到先得。所以每个等待线程订阅的都是不同的channel:redisson_lock__channel:{myLock}:UUID:threadId。当某个线程释放锁的时候,只会往等待队列中第一个线程对应订阅的channel发送消息。
最后,返回1
场景:
lua脚本:
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; "
分析:
hexists myLock UUID:threadId
场景:
lua脚本:
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "end; "
分析:
利用 hincrby 扣减当前线程的加锁次数
hincrby myLock UUID:threadId -1
如果扣减后次数还是大于0,证明是重复获取锁,所以此时只需要重新刷新锁的过期时间,然后返回0
expire myLock 30_000
场景:
lua脚本:
"redis.call('del', KEYS[1]); " + "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + "if nextThreadId ~= false then " + "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " + "end; " + "return 1; "
分析:
利用 del 命令删除锁对应 redis key
del myLock
往等待线程的订阅channel发送消息,通知其可以尝试获取锁了
lindex redisson_lock_queue:{myLock} 0
publish redisson_lock__channel:{myLock}:UUID:threadId 0
接着调用释放锁后续的操作:例如停止 watchdog 运行、执行 tryFailure 或 trySuccess 等和 RedissonClient 保持一致,这里就不再详细介绍了~