3.Redisson源码，公平锁排队分数刷新
2020-10-25 本文已影响0人
T_log
- 客户端B和客户端C都是处于排队的状态，他们在java源码中，会进入while ture循环，每次等待一段时间之后，就会重新来尝试进行加锁，客户端B和客户端C，模拟他们等待一段时间之后，来尝试进行重新加锁
- 客户端B，第一次尝试加锁时间假设是10:00:00，假设当前时间是10:00:15，他来再次尝试进行加锁，但是客户端A此时还是持有着这把锁的，此时会出现什么情况呢？
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）
代码片段一、RedissonFairLock
这里的1、2相当于之前笔记中的客户端B和客户端C，参考之前的笔记哦
@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
"while true do “
// 1. 获取队列中redisson_lock_queue:{anyLock}第一个元素的值，此时firstThread2 = UUID_02:threadId_02
// 2. 第三个(结果1笔记)客户端C，再次尝试加锁 尝试时间为：10:00:20
+ "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);"
+ "if firstThreadId2 == false then "
+ "break;"
+ "end; “
// 1. 获取redisson_lock_timeout:{anyLock} UUID_02:threadId_02的timeout时间 10:00:25(客户端A) <= 10:00:30 ？条件不成立，跳出循环
// 2. 10:00:30 < 10:00:20? 不成立
+ "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;"
// 1. 判断有没有人加锁
+ "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; “ +
// 1. 以及判断是否是客户端B加锁，都不成立
"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. 从队列redisson_lock_queue:{anyLock}取出第一个元素，firstThreadId = UUID_02:threadId_02
// 2. 第三个客户端C过来，从队列redisson_lock_queue:{anyLock}取出第一个元素，firstThreadId = UUID_02:threadId_02
"local firstThreadId = redis.call('lindex', KEYS[2], 0); " +
"local ttl; “ +
// 1. firstThreadId 不等于空 并且 firstThreadId 不等于 UUID_02:threadId_02,条件不成立
// 2. firstThreadId 不等于空 并且 firstThreadId 不等于 UUID_02:threadId_02,条件成立
"if firstThreadId ~= false and firstThreadId ~= ARGV[2] then “ +
// ttl = 10:00:30(这个应该是客户端B的超时时间) - 10:00:20(当前客户端再次获取锁的时间) = 10000毫秒
"ttl = tonumber(redis.call('zscore', KEYS[3], firstThreadId)) - tonumber(ARGV[4]);" +
"else “
// 1. 读取客户端A获取锁的生存周期 10:00:25(客户端A的超时时间参考笔记1) - 10:00:15(当前时间) = 10000毫秒
+ "ttl = redis.call('pttl', KEYS[1]);" +
"end; " +
// 1. timeout = 10000 + 10:00:15 + 5000 = 10:00:30
// 2.timeout = 1000毫秒 + 10:00:20 + 5000=10：00：35
"local timeout = ttl + tonumber(ARGV[3]);” +
// 1. zadd redisson_lock_timeout 10:00:30 UUID_02:threadId_02,刷新一下有序集合中的元素的分数，10:00:30
// 1. 其实这里需要注意的是zadd指令，需要当前这个key不存在，则直接插入，返回1，如果已经存在，则只是刷新当前key对应元素的分数，所以他不会向队列中插入重复的元素
// 2. zadd redisson_lock_timeout 10:00:35 UUID_03:threadId_03,刷新一下有序集合中的元素的分数，10:00:35
"if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " +
"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();
}
我终于把我的全部献给了程序
