Redisson 实现分布式锁源码浅析

代码示例

@Slf4j
public class WatchDogDemo {

    static Config config = null;
    static Redisson redisson = null;
    static final String KEY_WATCH = "watch_dog_key";
    static{
        config = new Config();
        config.useSingleServer().setAddress("redis://localhost:6379").setDatabase(0);
        redisson = (Redisson)Redisson.create();
    }

    public static void main(String[] args) {
        RLock lock = redisson.getLock(KEY_WATCH);
        lock.lock();
        try {
            log.info(">>>>>>>>11111.......abc");
            TimeUnit.SECONDS.sleep(25);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

流程解析

直接从 lock() 和 unlock() 看起，首先跟踪进入到 lock() 方法核心，如下：

继续进入到核心方法，如下：

如果自己不设置过期时间，那么底层默认会给你设置 30s 作为锁过期时间，如下：

进入到 tryLockInnerAsync() 加锁的方法如下：

会发现就是通过三段 lua 脚本指令去加锁，其中 KEYS[1] 代表是你加锁的那个 key，ARGV[1] 代表过期时间(默认30s)、ARGV[2] 代表加锁的客户端 ID。

第一段 lua 脚本表示： 第一个客户端过来判断是否存锁 key 是否已经存在，第一次过来很明显不存在这个 key，所以需要去上锁，从而执行 hincrby 指令开始加锁，并且给这个客户端分配了一个唯一标识的客户端 ID (ARGV[2]) ，并且通过expire设置过期时间为 30s，然后最后返回 null。

第二段 lua 脚本表示：假如第二次过来的还是第一个客户端，判断已经存在锁了，那么就要去执行锁重入操作，给这个客户端的锁次数 +1，在释放锁的时候也要做相应的 -1 操作，然后最后返回 null。

第三段 lua 脚本表示：假设第二次过来的是另一个客户端，会先执行第一段判断第二个客户端是否要加锁，假设第一个客户端还没有释放掉锁，那么第二个客户端就不能加锁成功，只能去执行第三段 lua 指令，然后最后返回第一个客户端持有锁的过期时间，注意前两步都是返回的 null，只有最后一步是返回持有锁的过期时间。

可以总结出一句话，分布式锁的加锁和设置过期时间都是通过三段 lua 脚本指令实现的，一条指令的执行具备原子性。

接着继续分析下面一段代码，如下：

通过 tryLockInnerAsync() 方法已经可以尝试加锁：

第一种情况：

如果加锁成功，然后 lua 脚本直接就会返回 null，所以 ttlRemaining 变量值就是 null，然后进入 if 逻辑。leaseTime 默认传入的值是 -1。所以走 else 分支的逻辑。

重点看这个 scheduleExpirationRenewal() 方法，如下：

首先能够看到这是一个递归调用，自己调用自己，递归调用最容易出现的就是栈溢出问题，所以得给一个休息的时间，所以每隔 1/3 时间才会去执行一次，默认时间 30s，所以就是每隔 10s 执行一次，主要去干这样一件事情，重新刷新锁的过期时间。

把这段代码简化成自己能看到的代码如下所示：

然后主要是在刷新锁的过期时间(虽然已加锁成功就开启了 watchdog 监控🐶，但是并不是一开始就去刷新时间，而是过了 10s 才会去刷新这个过期时间)。进入 renewExpirationAsync() 方法如下：

会发现也是两段 lua 脚本指令：

第一段会去判断你这个锁是否还被人持有着，只要是这把锁还在被人使用，那么肯定不能让这把锁过期失效，所以要给这个把锁续命，或者说重新刷新回默认的 30s，ARGV[1] 就是表示一开始赋值的 30s 时间。最后一定要非常注意返回的是 1。

第二段自然就是表示锁没有人使用了呗，那么肯定就不用去做续命了，直接返回 0。

然后继续往下看代码，如下：

如果 lua 返回 1，res 就是 true，如果 lua 返回 0，res 就是 false，返回 1 表示需要续命，那就会在此调用自己 renewExpiration() 方法，然后又去执行这段 lua 脚本重新刷新锁过期时间。

什么时候能够跳出呢？当锁不需要续命(过期或者释放了)的时候，那么这里就会返回 0，走 else 逻辑就可以跳出了，自己不去调用自己了不就可以退出了么。

第二种情况：

如果加锁不成功，然后 lua 脚本直接就会返回这个锁还有多少时间过期，所以 ttlRemaining 变量值不为 null。直接走 return CompletableFutureWrapper(f) 逻辑

然后继续观察代码如下：

tryAcquire(-1, leaseTime, unit, threadId) 表示继续尝试去加锁是否能成功，假设还是失败的，继续看下图：

ttl 返回值大于0 表示加锁失败，那么就要让这个线程不要这么频繁的去尝试了，等人家通知你吧，这里就使用到了 AQS 里面闭锁 Semaphore 信号量作为拦截，信号大小初始设置成 0 了，表示要让这个线程去睡眠了

最终会调用 park() 方法线程被挂起，等着被唤醒(等着锁释放或者其他线程被中断了就会醒来)

最后再来看下锁释放的源码如下：

主要干两件事情，一是去释放锁肯定是去执行 lua 脚本指令，毕竟加锁都是使用的 lua，二是去取消 watchdog 的锁续命操作。

最后看下这个解锁的 lua 脚本如下：

也是分成对应的三段：

第一段是：如果发现当前线程和加锁线程不是同一个线程，不允许解锁，直接返回 null，解锁失败。

第二段是：每次调用 hincrby 递减 1，释放一次锁，如果发现剩余次数还大于 0 ，表示是自增锁，那就在把过期时间重新刷新下。

第三段是：调用 del 删除锁，并且还发布了删除消息，返回 1 表示释放锁成功。