一、说明
- 其实公平锁释放的源码也在RedissonFairLock中，unlockInnerAsync，笔记是接着之前的笔记一起的，所以需要连续的看下来
二、源码中的参数
这里的参数和获取锁的时候大部分是一样的，但是有略微的不同，还需要注意
KEYS = Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName)
KEYS[1] = getName() = 锁的名字，“anyLock”
KEYS[2] = threadsQueueName = redisson_lock_queue:{anyLock}，基于redis的数据结构实现的一个队列
KEYS[3] = timeoutSetName = redisson_lock_timeout:{anyLock}，基于redis的数据结构实现的一个Set数据集合，有序集合，可以自动按照你给每个数据指定的一个分数（score）来进行排序
KEYS[4] = redisson_lock_channel:{anyLock}
ARGV = internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime
ARGV[1] = 0L
ARGV[2] = 30 * 1000
ARGV[3] = UUID:threadId
ARGV[4] = 当前时间（10:00:00）
三、代码
- 代码片段一、
@Override
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
// remove stale threads
// 1. 首先会进入到这个死循环里面
"while true do “
// 1. Lindex redisson_lock_queue:{anyLock} 0 取出队列的第一个元素
+ "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);”
// 1. 如果队列中不存在的话，直接跳出死循环
+ "if firstThreadId2 == false then "
+ "break;"
+ "end; “
// 1. 如果队列的第一个元素存在的话，则zscore redisson_lock_timeout:{anyLock} 取出他的超时时间
+ "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));”
// 1. 如果超时时间timeout <= 当前时间，说明已经超时了
+ "if timeout <= tonumber(ARGV[4]) then “
// 1. zerm redisson_lock_timeout:{anyLock} UUID_01:threadId_01,删除这个客户端，其实这里的01代表的是当前客户端的UUID和线程ID
+ "redis.call('zrem', KEYS[3], firstThreadId2); “
// 1. lpop redisson_lock_queue:{anyLock} 从队列中删除这个元素
+ "redis.call('lpop', KEYS[2]); "
+ "else "
+ "break;"
+ "end; "
+ "end;"
// 1. exists anyLock==0 如果anyLock不存在的话
+ "if (redis.call('exists', KEYS[1]) == 0) then “ +
// 1. lindex redisson_lock_queue:{anyLock} 0 取出队列中的第一个元素
"local nextThreadId = redis.call('lindex', KEYS[2], 0); “ +
// 1. 如果队列中的元素也不存在，说明这个队列现在是空的
"if nextThreadId ~= false then “ +
// 1. 发布一个订阅事件，具体的订阅事件是干嘛的，暂时还看不出来
"redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
"end; " +
"return 1; " +
"end;” +
// 1. hexists anyLock 当前时间（10:00:00） + 5000毫秒 = 10:00:05
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; “ +
// 1.hincrby anyLock 当前时间（10:00:00） + 5000毫秒 = 10:00:05 减一，其实意思就是如果是可重入加锁的话，减去一
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); “ +
// 1. 如果加锁次数还是大于0的话
"if (counter > 0) then “ +
//1.pexpire anyLock UUID_01:threadId_01 更新他的生存周期
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"end; " +
// 1. 删除anyLock,如果走到这一步，说明是当前持有锁的客户端释放锁，那么这个时候，释放锁，然后从队列中取出第一个元素，发布一个订阅事件，让后面的排队的线程或者客户端尝试获取锁
"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.<Object>asList(getName(), threadsQueueName, timeoutSetName, getChannelName()),
LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis());
}
四、总结
- 在客户端A他释放锁的时候，也会走while true的脚本逻辑，看一下有序集合中的元素的timeout时间如果小于了当前时间，如果小于的话，就认为他的那个排队就过期了，就删除他，让他后面重新尝试获取锁的时候重排序
- while true的逻辑，比如说客户端B或者客户端C，他们用的是tryAcquire()方法，他们其实设置了一个获取锁超时的时间，比如说他们在队列里排队，但是尝试获取锁超过了20秒，人家就不再尝试获取锁了
- 此时他们还是在队列redisson_lock_queue{anyLock}和有序集合redisson_lock_timeout:{anyLock}里占了一个坑位，while true的逻辑就可以保证说剔除掉这种不再尝试获取锁的客户端，有序集合里的timeout分数就不会刷新了，随着时间的推移，肯定就会剔除掉他
- 如果客户端宕机了，也会导致他就不会重新尝试来获取锁，也就不会刷新有序集合中的timeout分数，不会延长timeout分数，while true的逻辑也可以剔除掉这种宕机的客户端在队列里的占用
- 因为网络延迟等各种因素在里面，可能会在等待锁时间过长的时候，触发各个客户端的排队的顺序的重排序，有的客户端如果在队列里等待时间过长了，那么其实是可以触发一次队列的重排序的
- 他在这里发布一个锁被释放的消息，肯定在他的源码中是有一些人是订阅了这个释放锁的消息的，此时他们就可以得到一个锁被释放掉的通知
五、寄语
每天都会有很多不一样有意思的想法和灵感，想什么很重要，用行动去实践自己的想法也很重要，不管对的错的，走过了，经历了，才知道