公平锁加锁的源码在RedissonFairLock
对于lua脚本中的一些参数值的说明,因为lua脚本中设计到很多的参数,提交的提取出来,方便看lua脚本的时候进行分析
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:00) + 5000毫秒 = 10:00:05
ARGV[4] = 当前时间(10:00:00)
lua脚本中的1、2、3 分别代表第一个、第二个、第三个客户端,代表加锁的顺序
@Override
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
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);
}
// 公平锁源码入口
if (command == RedisCommands.EVAL_LONG) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
// remove stale threads
// 1. 第一次,如果某个锁,没有人加锁,此时第一个客户端进来,先进入一个死循环
// 2. 第二个客户端进来尝试加锁,进入死循环中
// 3. 第三个客户端过来加锁,进入while true 的死循环当中
"while true do “
// 1. 第一次进来,这个lindex redisson_lock_queue:{anyLock} 0 的命令就是说,从 redisson_lock_queue:{lock}这个队列中弹出第一个元素,刚开始,肯定是空的,什么都没有,直接就会直接
// break掉,跳出死循环
// 2. 第二个客户端首先执行 lindex redisson_lock_queue:{anyLock} 0,取出队列的第一个元素,此时该队列还是空的,然后break掉,跳出死循环
// 3. lindex redisson_lock_queue:{anyLock} 0,取出队列的第一个元素,此时,这个队列中由于第二个客户端已经将数据插入到队列中,已经在排队了,所以值是有的firstThreadId2=10:00:25
+ "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);"
+ "if firstThreadId2 == false then "
+ "break;"
+ "end; “
// 3. zscore redisson_lock_timeout:{anyLock} UUID_02:threadId_02,从有序集合中获取UUID_02:threadId_02对应的分数10:00:25
+ "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));”
// 3. 如果第三个客户端尝试加锁的时间是10:00:05,timeout < 10:00:05,如果条件成立,跳出死循环,如果不成立,具体逻辑后面的笔记中进行分析
+ "if timeout <= tonumber(ARGV[4]) then "
+ "redis.call('zrem', KEYS[3], firstThreadId2); "
+ "redis.call('lpop', KEYS[2]); "
+ "else "
+ "break;"
+ "end; "
+ "end;"
// 1. exists anyLock,锁不存在,也就是没人加锁,刚开始因为第一个客户端进来,之前肯定是没有人加锁的,这个条件是成立的
// 1. exists redisson_lock_queue:{anyLock}这个队列不存在,或者lindex redisson_lock_queue:{anyLock} 0 这个队列中的第一个元素是UUID:threadId ,
// 这个队列存在,但是排在这个队列的第一个元素是当前线程,那么此时这个条件就会成立。
// 第一个客户端进来的时候,其实anyLock和队列redisson_lock_queue:{anyLock}都是不存在的,所以条件是成立的
// 2. exists anyLock 是否存在,这个时候肯定已经存在了,所以这里的if 判断条件不成立
// 3. 第三个客户端进行条件判断,和第二个客户端是一样的,条件不成立
+ "if (redis.call('exists', KEYS[1]) == 0) and ((redis.call('exists', KEYS[2]) == 0) "
+ "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then “ +
// 1. lpop redisson_lock_queue:{anylock} 弹出第一个元素,,队列是空的,所以第一个客户端进来的时候,这行命令是什么都不会干的
"redis.call('lpop', KEYS[2]); “ +
// 1. zrem redisson_lock_timeout:{anyLock} UUID_01:threadId,从set集合中删除threadId对应的元素,此时set集合是空的,所以什么都不会做
"redis.call('zrem', KEYS[3], ARGV[2]); “ +
// 1. hset anyLock uuid:threadId 1 进行加锁,数据结构就是 anyLock:{“UUID_01:threadId_01” : 1 }
"redis.call('hset', KEYS[1], ARGV[2], 1); “ +
// 1. Pexpire anyLock 30000 ,将anyLock这个key的生存周期设置为3000毫秒(30秒)
"redis.call('pexpire', KEYS[1], ARGV[1]); “ +
// 1. 返回一个nil,在外层代码中,就会认为是加锁成功,此时就会开启一个watchdog看门狗定时调度的程序,
// 每隔10秒判断一下,当前这个线程是否还对这个锁key持有着锁,如果是,则刷新锁key的生存时间为30000毫秒
"return nil; " +
"end; “ +
// 2. 第二个客户端会进入到这段逻辑中,hexists anyLock UUID_02:threadId_02是否存在,条件也不成立
// 3. 第三个客户端进行这段逻辑,hexists anyLock UUID_03:threadId_03是否存在,条件也不成立
"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; " +
// 2. 第二个客户端会执行如下的逻辑,lindex redisson_lock_queue:{anyLock} 0 从队列中获取第一个元素,肯定是空的
// 3. 第三个客户端执行lindex redisson_lock_queue:{anyLock} 0,从队列中取出第一个元素,,这个是有值的,为UUID_02:thread_02
"local firstThreadId = redis.call('lindex', KEYS[2], 0); “ +
// 取当前的时间
"local ttl; “ +
// 2. 判断firstThreadId 不等于空and firstThreadId=UUID_02:threadId_02条件不成立,走else逻辑
// 3. 判断firstThreadId 不等于空and firstThreadId!=UUID_03:threadId_03条件成立,条件是成立的,其实就是判断在队列中排队的第一个元素是不是当前客户端
"if firstThreadId ~= false and firstThreadId ~= ARGV[2] then “ +
// 3. Zscore redisson_lock_timeout:{anyLock} UUID_02:threadId_02 = 10:00:25 - 10:00:05 = 20000毫秒
"ttl = tonumber(redis.call('zscore', KEYS[3], firstThreadId)) - tonumber(ARGV[4]);" +
"else “
// 2. ttl = pttl anyLock,为anyLock的剩余生存时间,假设当前剩余20000毫秒
// 3. ttl = 20000毫秒
+ "ttl = redis.call('pttl', KEYS[1]);" +
"end; “ +
// 假设当前时间为10:00:00
// 2. timeout = ttl + 当前时间 + 50000 = 20000 + :10:00:00 + 5000 = 10:00:25
// 3. timeout = 20000毫秒 + 10:00:05 + 5000毫秒 = 10:00:30
"local timeout = ttl + tonumber(ARGV[3]);” +
// 2. zadd rediss_lock_timeout:{anyLock} 10:00:25 UUID_02:threadId_02,在set集合中插入一个元素,元素的值是UUID_02:threadId_02
// 2. 他对应的分数应该是10:00:25(会用这个时间对应的时间戳,也就是long类型表示这个时间,时间越靠后值就越大),而sort set是一个有序集合,会
// 2. 会根据这个分数进行排序
// 3.zadd redisson_lock_timeout:{anyLock} 10:00:30 UUID_03:threadId_03
"if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then “ +
// 2. rpush redisson_lock_queue:{anyLock} UUID_02:threadId_02 ,将UUID_02:threadId02插入到这个队列中
// 3. rpush redisson_lock_queue:{anyLock} UUID_03:theadId_03
"redis.call('rpush', KEYS[2], ARGV[2]);" +
"end; “ +
// 2. 返回ttl,就是anyLock的剩余生存时间,如果拿到的ttl是一个数字的话,那么第二个客户端就会进入死循环,每隔一段时间过来进行尝试加锁,重新执行这段lua脚本
"return ttl;",
Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
internalLockLeaseTime, getLockName(threadId), currentTime + threadWaitTime, currentTime);
}
throw new IllegalArgumentException();
}