一、说明
- 笔记在一点点的更新,对于公平锁的加锁,其实Lua脚本都是同一套,我们在分析源码的时候,都是假设当前时间,其实这个也很重要的,假设的条件也是根据实际情况。
- 如果客户端A释放了锁,删除了锁key之后,客户端B和客户端C是如何按照顺序依次加锁的。假设客户端B在队列里面是排在后面的,假设锁被释放掉了之后,是客户端B先来尝试加锁
- 假设当前时间是10:00:40,锁已经被释放了,客户端B来尝试重新加锁,只会刷新客户端B在有序集合中的分数,获取锁是失败的
- 那么如果B获取失败了,C排队队头,那么此时,如果客户端C来尝试获取锁的情况呢?源码中的2代表客户端C获取锁,1代表客户端B获取锁
二、源码中的参数
- 单独取出源码的参数,方便阅读源码
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)来进行排序
ARGV = internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime
ARGV[1] = 30000毫秒
ARGV[2] = UUID:threadId
ARGV[3] = 当前时间(10:00:40) + 5000毫秒 = 10:00:05
ARGV[4] = 当前时间(10:00:40)(这里的时间和之前笔记中的时间已经向后移动了)
三、源码
RedissonFairLock类中
@Override
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
//********************BEGIN-command为EVAL_LONG,所以不走这段逻辑*************************
long currentTime = System.currentTimeMillis();
if (command == RedisCommands.EVAL_NULL_BOOLEAN) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
// 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[3]) then "
+ "redis.call('zrem', KEYS[3], firstThreadId2); "
+ "redis.call('lpop', KEYS[2]); "
+ "else "
+ "break;"
+ "end; "
+ "end;"
+
"if (redis.call('exists', KEYS[1]) == 0) and ((redis.call('exists', KEYS[2]) == 0) "
+ "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +
"redis.call('lpop', KEYS[2]); " +
"redis.call('zrem', KEYS[3], ARGV[2]); " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return 1;",
Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
internalLockLeaseTime, getLockName(threadId), currentTime);
}
////********************END-command为EVAL_LONG,所以不走这段逻辑*************************
if (command == RedisCommands.EVAL_LONG) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
// remove stale threads
// 1.客户端B进入到死循环逻辑
"while true do “
// 1.lindex redisson_lock_queue:{anyLock} 0 取出队列中的第一个元素,
// 1. 此时队列的第一个元素是客户端C,firstThreadId2 = UUID_03:threadId_03
+ "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);”
// 1. 因为firstThreadId2不为空,所以if条件不成立
+ "if firstThreadId2 == false then "
+ "break;"
+ "end; “
// 1.zscore redisson_lock_timeout:{anyLock} UUID_03:threadId_03,取出队列中的第一个元素的分数
+ "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));”
// 1. 10:01:04(客户端C的超时时间) <= 10:00:40(当前时间) 条件不成立
+ "if timeout <= tonumber(ARGV[4]) then "
+ "redis.call('zrem', KEYS[3], firstThreadId2); "
+ "redis.call('lpop', KEYS[2]); "
+ "else "
+ "break;"
+ "end; "
+ "end;"
// 1.exists anyLock == 0 锁不存在,这个条件是成立的,因为客户端A已经释放锁了
// 1.exists redisson_lock_queue:{anyLock} 不存在或者lindex redisson_lock_queue:{anyLock} 第一个元素的值
// 1. UUID_03:threadId_03 = UUID_02:threadId_02(当前是客户端B来获取锁),所以综合,if条件不成立,无法加锁
// 2.anyLock 不存在,而且队头就是客户端C,所以if条件成立,进入正常的加锁逻辑
+ "if (redis.call('exists', KEYS[1]) == 0) and ((redis.call('exists', KEYS[2]) == 0) "
+ "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then “ +
//2.lpop redisson_lock_queue:{anyLock} 将客户端C从队列中弹出
//2.完成加锁,而且客户端C从队列中出队,此时排在队头的就是客户端B了
"redis.call('lpop', KEYS[2]); “ +
//2.zerm redisson_lock_timeout:{anyLock} UUID_03:threadId_03 ,从有序集合中把客户端C的线程ID删除
"redis.call('zrem', KEYS[3], ARGV[2]); “ +
// 2.hset anyLock UUID_03:threadId_03 1
// 2.数据结构为:anyLock:{UUID_03:threadiD_03 : 1}其实就是对客户端C进行加锁
"redis.call('hset', KEYS[1], ARGV[2], 1); “ +
// 2. pexpire anyLock 30000 刷新锁的生存时间为30000毫秒,即:30秒
"redis.call('pexpire', KEYS[1], ARGV[1]); “ +
// 2.返回nil,java代码拿到null,就会启动watchdog看门狗,后台定时调度定时进行刷新
"return nil; " +
"end; “ +
// 1.hexists anyLock UUID_03:threadId_03 == 1,其实仔细想一下就知道了,这块的anyLock已经不存在了,
// 所以这里的if条件也是不成立的,其实稍微想一下,就知道,客户端A释放锁的时候就已经把anyLock给删除了,
// 1.这里的条件是怎么都不会成立的
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
//1.lindex redisson_lock_queue:{anyLock} 0 取出队列的第一个元素,UUID_03:threadId_03,为客户端C
"local firstThreadId = redis.call('lindex', KEYS[2], 0); " +
"local ttl; “ +
// 1.if条件成立,首先是firstThreadId不为空且firstThreadId不等于UUID_03:threadId_03
"if firstThreadId ~= false and firstThreadId ~= ARGV[2] then “ +
// 1.重新计算ttl,zscore redisson_lock_timeout:{anyLock} UUID_03:threadId_03
// 1. 10:01:04(客户端C的超时时间) - 10:00:40(当前时间) = 24000毫秒
"ttl = tonumber(redis.call('zscore', KEYS[3], firstThreadId)) - tonumber(ARGV[4]);" +
"else “
+ "ttl = redis.call('pttl', KEYS[1]);" +
"end; " +
// 1. timeout = 24000毫秒 + 10:00:40 + 5000 = 10:01:09(在Redis里面的数据表现形式是long类型的时间戳)
"local timeout = ttl + tonumber(ARGV[3]);” +
// 1.zadd redisson_lock_timeout:{anyLock} 10:01:09 UUID_02:threadId_02
// 1. 刷新一下客户端B在有序集合中的timeout分数,10:01:09
// 1. 所以,根据我们之前的假设,即使客户端A释放了锁,但是现在排在队头的是客户端C,
// 1.客户端B现在来获取锁也是获取不到的,只是把客户端B在有序集合中的分数刷新一下而已
"if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then “ +
// 1. rpush redisson_lock_queue:{anyLock} UUID_02:threadId_02
"redis.call('rpush', KEYS[2], ARGV[2]);" +
"end; " +
"return ttl;",
Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime);
}
throw new IllegalArgumentException();
}
四、总结
- 获取锁超时,其他客户端获取不到锁,一定会在java代码里进入一个while true死循环,一定时间内没有获取到锁,就返回false标识获取锁失败,过了一段时间,只要没有刷新有序集合中的timeout分数,就会自然被lua脚本里的while true逻辑给清理掉
- 超时自动释放锁,不会开启lock watchdog后台定时调度的任务