一、RedissonFairLock#unlock 源码分析

上一篇讲到，RedissonFairLock 它主要是基于 RedissonLock 做的扩展，主要扩展在于加锁和释放锁的地方。所以我们要研究 RedissonFairLock 释放锁的逻辑，只需要关注它重写的 RedissonLock#unlockAsync 方法即可。至于如何计算 slot 来执行 lua 脚本，释放后需停止 wathchdog 的运行，和 RedissonLock 保持一致，我们只要研究了 RedissonLock 释放锁的逻辑后，这里就不再需要关注了。那我们下面直接上lua脚本的分析。

1、RedissonFairLock 之 lua 脚本释放锁

RedissonFairLock#unlockInnerAsync：

@Override
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            // 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[4]) then "
                + "redis.call('zrem', KEYS[3], firstThreadId2); "
                + "redis.call('lpop', KEYS[2]); "
            + "else "
                + "break;"
            + "end; "
          + "end;"

          + "if (redis.call('exists', KEYS[1]) == 0) then " + 
                "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + 
                "if nextThreadId ~= false then " +
                    "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
                "end; " +
                "return 1; " +
            "end;" +
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
            "if (counter > 0) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
            "end; " +

            "redis.call('del', KEYS[1]); " +
            "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + 
            "if nextThreadId ~= false then " +
                "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
            "end; " +
            "return 1; ",
            Arrays.asList(getName(), threadsQueueName, timeoutSetName, getChannelName()),
            LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis());
}

因为基于 RedissonLock 加入了公平获取锁的机制，所以lua脚本还是挺长的，下面我们一步一步分析。

1.1、KEYS

Arrays.asList(getName(), threadsQueueName, timeoutSetName, getChannelName())：

• getName(): 锁key
• threadsQueueName：prefixName("redisson_lock_queue", name)，用于锁排队
• timeoutSetName：prefixName("redisson_lock_timeout", name)，用于队列中每个客户端的等待超时时间
• getChannelName：prefixName("redisson_lock__channel", getName())，当前锁对应的订阅channel

KEYS：["myLock","redisson_lock_queue:{myLock}","redisson_lock_timeout:{myLock}","redisson_lock__channel:{myLock}"]

1.2、ARGVS

LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis()：

• LockPubSub.UNLOCK_MESSAGE：0L
• internalLockLeaseTime：30_000毫秒，watchdog的超时时间
• getLockName(threadId)：return id + ":" + threadId，客户端ID(UUID):线程ID(threadId)
• System.currentTimeMillis()：当前时间时间戳

ARGVS：[0L,3w毫秒,"UUID:threadId",当前时间时间戳]

1.3、lua 脚本分析

1、分支一：清理过期的等待线程

场景：

和获取锁的第一步一样，开个死循环清理过期的等待线程，主要避免下面场景，避免无效客户端占用等待队列资源

• 获取锁失败，然后进入等待队列，但是网络出现问题，那么后续很有可能就不能继续正常获取锁了。
• 获取锁失败，然后进入等待队列，但是之后客户端所在服务器宕机了。

lua脚本：

"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[4]) then " +
        // remove the item from the queue and timeout set
        // NOTE we do not alter any other timeout
        "redis.call('zrem', KEYS[3], firstThreadId2);" +
        "redis.call('lpop', KEYS[2]);" +
    "else " +
        "break;" +
    "end;" +
"end;" +

分析：

1. 开启死循环
2. 利用 lindex 命令判断等待队列中第一个元素是否存在，如果存在，直接跳出循环

  lindex redisson_lock_queue:{myLock} 0

3. 如果等待队列中第一个元素不为空（例如返回了LockName，即客户端UUID拼接线程ID），利用 zscore 在 超时记录集合(sorted set) 中获取对应的超时时间

 zscore redisson_lock_timeout:{myLock} UUID:threadId

4. 如果超时时间已经小于当前时间，那么首先从超时集合中移除该节点，接着也在等待队列中弹出第一个节点

 zrem redisson_lock_timeout:{myLock} UUID:threadId
lpop redisson_lock_queue:{myLock}

5. 如果等待队列中的第一个元素还未超时，直接退出死循环

2、分支二：锁已经被释放，通知等待队列中第一个线程

场景：

• 成功获取锁线程重复调用释放锁的方法，第二次释放时，锁已不存在，就去通知等待队列中的第一个元素
• 又或者一个极端场景：当前线程未能成功获取锁，但是调用了释放锁的方法，并且刚好此时锁被释放

lua脚本：

"if (redis.call('exists', KEYS[1]) == 0) then " + 
    "local nextThreadId = redis.call('lindex', KEYS[2], 0); " + 
    "if nextThreadId ~= false then " +
        "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
    "end; " +
    "return 1; " +
"end;"

分析：

1. 利用 exists 命令判断锁是否存在

  exists myLock

2. 如果锁不存在，利用 lidex 获取等待队列中的第一个元素

lindex redisson_lock_queue:{myLock} 0

3. 如果等待列表中第一个元素不为空，即还存在等待线程，往等待线程的订阅channel发送消息，通知其可以尝试获取锁了

  lindex redisson_lock_queue:{myLock} 0

关于这个订阅通道，之前介绍 RedissonLock 死循环等待锁的时候没有详细介绍，这个可以补充一下。RedissonLock 所有等待线程都是订阅锁的同一个channel：redisson_lock__channel:{myLock}，当有线程释放锁的时候，会往这个通道发送消息，此时所有等待线程都可以订阅消费这条消息，从而从等待状态中释放出来，重新尝试获取锁。

而 RedissonFairLock 不太一样，因为它要支持公平获取锁，即先到先得。所以每个等待线程订阅的都是不同的channel：redisson_lock__channel:{myLock}:UUID:threadId。当某个线程释放锁的时候，只会往等待队列中第一个线程对应订阅的channel发送消息。 4. 最后，返回1

3、分支三：加锁记录中的线程不是当前线程

场景：

• 当前线程未能成功获取锁，但是调用了释放锁的方法

lua脚本：

"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
    "return nil;" +
"end; "

分析：

1. 利用 hexists 命令判断加锁记录集合中，是否存在当前客户端当前线程

  hexists myLock UUID:threadId

2. 加锁记录不存在当前线程，返回nil 如果返回null，会打印相关日志，并调用 tryFailure 方法。

4、分支四：当前线程拥有锁，并且获取锁次数大于1

场景：

• 当前线程拥有锁，并且重入次数大于1

lua脚本：

"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
    "return 0; " +
"end; "

分析：

1. 利用 hincrby 扣减当前线程的加锁次数

 hincrby myLock UUID:threadId -1

2. 如果扣减后次数还是大于0，证明是重复获取锁，所以此时只需要重新刷新锁的过期时间，然后返回0

 expire myLock 30000

5、分支五：当前线程成功释放锁

场景：

• 如果前面的四个分支都不符合，证明当前线程持有锁，并且只加锁一次
• 这里只需要删除锁key，然后通知等待队列中的第一个线程即可

lua脚本：

"redis.call('del', KEYS[1]); " +
"local nextThreadId = redis.call('lindex', KEYS[2], 0); " + 
"if nextThreadId ~= false then " +
    "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
"end; " +
"return 1; "

分析：

1. 利用 del 命令删除锁对应 redis key

  del myLock

2. 往等待线程的订阅channel发送消息，通知其可以尝试获取锁了

  lindex redisson_lock_queue:{myLock} 0

 publish redisson_lock__channel:{myLock}:UUID:threadId 0

• 利用订阅channel给等待线程发送通知

• 利用 lindex 获取等待队列中的第一个线程

1.4 后续

接着调用释放锁后续的操作：例如停止 watchdog 运行、执行 tryFailure 或 trySuccess 等和 RedissonClient 保持一致，这里就不再详细介绍了～

本文由哈喽比特于2年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/IrtLuRZ2cxlfVw_zmNLjGA