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

6. Redisson源码剖析-公平锁按顺序依次加锁

一、说明

  1. 笔记在一点点的更新,对于公平锁的加锁,其实Lua脚本都是同一套,我们在分析源码的时候,都是假设当前时间,其实这个也很重要的,假设的条件也是根据实际情况。
  2. 如果客户端A释放了锁,删除了锁key之后,客户端B和客户端C是如何按照顺序依次加锁的。假设客户端B在队列里面是排在后面的,假设锁被释放掉了之后,是客户端B先来尝试加锁
  3. 假设当前时间是10:00:40,锁已经被释放了,客户端B来尝试重新加锁,只会刷新客户端B在有序集合中的分数,获取锁是失败的
  4. 那么如果B获取失败了,C排队队头,那么此时,如果客户端C来尝试获取锁的情况呢?源码中的2代表客户端C获取锁,1代表客户端B获取锁

二、源码中的参数

  1. 单独取出源码的参数,方便阅读源码
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();
}

四、总结

  1. 获取锁超时,其他客户端获取不到锁,一定会在java代码里进入一个while true死循环,一定时间内没有获取到锁,就返回false标识获取锁失败,过了一段时间,只要没有刷新有序集合中的timeout分数,就会自然被lua脚本里的while true逻辑给清理掉
  2. 超时自动释放锁,不会开启lock watchdog后台定时调度的任务

https://www.xamrdz.com/backend/3g61942647.html

相关文章: