当前位置: 首页>后端>正文

3.Redisson源码,公平锁排队分数刷新

  1. 客户端B和客户端C都是处于排队的状态,他们在java源码中,会进入while ture循环,每次等待一段时间之后,就会重新来尝试进行加锁,客户端B和客户端C,模拟他们等待一段时间之后,来尝试进行重新加锁
  2. 客户端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();
}

我终于把我的全部献给了程序


https://www.xamrdz.com/backend/33s1937909.html

相关文章: