分布式锁之Redisson实现

桃扇骨 2022-12-18 10:54 94阅读 0赞

分布式锁相信大家都已经听过了，常见的方案呢，也就那么几种，今天我们来讲讲使用Redisson框架来实现redis的分布式锁

那么第一个问题来了，为什么不直接使用redis，而是要来用Redisson框架呢？

如果我们要使用redis来实现分布式锁的话，最low的一种方式就是直接set一个key，如果set成功了，那么就相当于持有了这把锁，其他的线程无法set成功，就只能不断的轮询尝试获取锁，这就是最基础的redis分布式锁原理

设想一下以下的问题：

1.  当一个线程加锁成功了，然后他自己给挂了，那么之后的线程是不是永远也无法获取到这个分布式锁了？聪明的你肯定想到了，我们可以设置一个过期时间，那设置过期时间和设置key是不是一个原子性的命令呢？如果一个命令设置失败了那该如何解决？
2.  如果加锁的线程加锁的时间比较长，超过了我们设置的过期时间，那么超时之后就自动释放锁了，那么我们之前所做的操作该怎么办？
3.  在使用锁的时候，我们知道JUC包下有很多种不同的锁以供不同的场景来使用，只是使用上面说的最简单的可重入锁，可能很多场景的并发量都会大幅降低，这是我们所不能接受的

接下来我们通过简单的源码分析来看看Redisson框架是如何帮助我们解决这些问题的，要找到源码的入口，首先我们得知道如何使用Redisson实现分布式锁

RLock lock = redisson.getLock("anyLock");
    // 最常见的使用方法
    lock.lock();

没错，就是这么简单，上面的anyLock就相当于你要加锁的key值，使用一个lock()方法就能实现加锁逻辑了，解锁跟加锁逻辑一样简单

lock.unlock();

好的，那么我们带着我们上面所提的三个问题，来探秘一下Redisson内部是如何解决的

只要顺着lock()方法，一步步跟进代码中，就能得到我们想要的答案，我就不一步一步跟进来了，这里直接贴出最终的逻辑

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
            internalLockLeaseTime = unit.toMillis(leaseTime);
    
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                      "if (redis.call('exists', KEYS[1]) == 0) then " +
                          "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 redis.call('pttl', KEYS[1]);",
                        Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
        }

我们可以看到，这里执行了一个commandExecutor.evalWriteAsync方法，里面感觉执行了一大段逻辑，其实这里就是Redisson最核心的加锁逻辑了，这是一个lua脚本，通过这个lua脚本，就能实现操作的原子性，这就解决了我们提到的第一个问题。**Redisson的加锁操作使用lua脚本来保证原子性**

那我们来分析这个方法的逻辑，首先这里有几个参数，KEYS\[1\]、ARGV\[1\]、ARGV\[2\]  
KEYS\[1\]的值就是 Collections.singletonList(getName())，也就是加锁的key的名称，还记得我们之前加锁的逻辑吗，这个值其实就是anyLock  
ARGV\[1\]的值是internalLockLeaseTime，这里的默认值是30秒  
ARGV\[2\]的值是getLockName(threadId)，这里可以理解成一个redis客户端的一个唯一线程，他是由一个UUID和一个threadId组成，我们这里给一个值来替代，例如8743c9c0-0795-4907-87fd-6c719a6b4586:1

我们假设第一个线程来执行加锁的逻辑，这时我们的逻辑是这一段代码

"if (redis.call('exists', KEYS[1]) == 0) then " +
                          "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                          "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                          "return nil; " +
                      "end; " +

第一步判断是否存在anyLock这个key值，如果不存在的话，就执行hset，设置了一个key，此时在redis中，就有这么一个数据结构

anyLock
    {
      “8743c9c0-0795-4907-87fd-6c719a6b4586:1”: 1
    }

接着执行pexpire命令，给anyLock设置了一个过期时间，默认是30秒，最后返回一个nil

如果是当前线程再次执行加锁逻辑会发生什么呢，我们知道这是一个可重入锁，之后就会执行下面这段代码

"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; " +

首先判断了之前的数据结构\{anyLock：8743c9c0-0795-4907-87fd-6c719a6b4586:1\} 是否存在，这在redis中是一个hash的数据结构，因为是同一个线程加锁，当然是存在的，之后可以很明显的看到，执行了一个hincrby的操作，这里就可以理解为重入加锁的次数，如果是第二次加锁的话，那么redis中的数据结果就变为了

anyLock
    {
      “8743c9c0-0795-4907-87fd-6c719a6b4586:1”: 2
    }

同时刷新了一下过期时间，最后返回一个nil

那么如果是其他线程来尝试获取这个分布式锁呢，第一步判断是否存在要加锁的key，我们知道已经有人加了锁，那么肯定存在，之后又走到第二段逻辑判断数据结构，因为是不同的线程，所以是不存在的，那么就直接执行lua脚本中的

return redis.call('pttl', KEYS[1]);

返回当前分布式锁的过期时间

整个加锁逻辑其实就这么点东西，还是比较好理解的

解锁的逻辑就稍微带一下吧

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                    "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "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; " +
                    "else " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; "+
                    "end; " +
                    "return nil;",
                    Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
        }

简单来说，如果是加锁线程解锁，会递减加锁的次数，直到为0，才会真正删除掉这个key

接下来我们来讲一个重头戏，也就是Redisson的watchdog机制，这个机制就是用来解决我们之前提到的第二个问题的  
在获取锁的逻辑中有这么一段代码

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        // 这里封装的其实是当前这个key对应的一个剩余的存活时间，单位是毫秒，pttl anyLock这个命令所返回的
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        // 给那个RFuture加了一个监听器，也就是说只要这个lua脚本执行完成，返回了pttl anyLock那个指令返回的一个剩余存活的时间之后
        // 这个RFuture的监听器就会被触发执行的
        ttlRemainingFuture.addListener(new FutureListener<Long>() {
            @Override
            public void operationComplete(Future<Long> future) throws Exception {
                // 如果那段加锁的lua脚本执行失败的话，那么这里就不是success，相当于是基于redis加锁失败了
                if (!future.isSuccess()) {
                    return;
                }
    
                Long ttlRemaining = future.getNow();
                // 正常情况下，ttlRemaining，也就是pttl那个指令返回的值，在这里呢其实是一个null
                // lock acquired
                if (ttlRemaining == null) {
                    // 如果锁是成功获取的话，接下来他就会执行一个逻辑，也就是后台开启一个定时调度的任务
                    // 只要这个锁还被客户端持有着，那么就不会不断的去延长那个key的生存周期
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }

scheduleExpirationRenewal()方法就是我们要找的watchdog的逻辑，我们进去仔细看看

private void scheduleExpirationRenewal(final long threadId) {
        // 判断expirationRenewalMap是否包含entryName
        if (expirationRenewalMap.containsKey(getEntryName())) {
            return;
        }
    
        // 开启一个调度任务
        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                // 执行下面的lua脚本，延长过期时间
                RFuture<Boolean> future = renewExpirationAsync(threadId);
                
                future.addListener(new FutureListener<Boolean>() {
                    @Override
                    public void operationComplete(Future<Boolean> future) throws Exception {
                        // 移除这个map的entryName
                        expirationRenewalMap.remove(getEntryName());
                        if (!future.isSuccess()) {
                            log.error("Can't update lock " + getName() + " expiration", future.cause());
                            return;
                        }
                        // 延迟过期成功之后，则再次设置scheduleExpirationRenewal也就是看门狗的任务
                        if (future.getNow()) {
                            // reschedule itself
                            scheduleExpirationRenewal(threadId);
                        }
                    }
                });
            }
    
        // internalLockLeaseTime默认为30秒，也就是默认10秒后调度一次看门狗的任务
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    
        // 使用putIfAbsent方法添加键值对，如果map集合中没有该key对应的值，则直接添加，并返回null，如果已经存在对应的值，则依旧为原来的值
        // putIfAbsent返回不为null的话，说明更新过期时间失败了，那么就直接取消任务
        if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
            task.cancel();
        }
    }

通过上面的逻辑，我们发现里面其实启动了一个TimerTask定时调度任务，默认的调用时间是internalLockLeaseTime / 3，也就是默认超时时间的三分之一，默认为10秒，每次延长超时时间成功之后又会再次调用这个任务，这么做的话，就可以实现一个自动续期的功能，即使一个线程长时间持有锁，也不会因为过期时间而释放锁了

**Redisson解决我们第二个问题的机制就是所谓的watchdog，通过看门狗机制，自动的去给过期时间进行续期，默认每隔超时时间的三分之一就去执行一次**

最后一个问题是需要多种不同类型的锁来适配多种不同的场景，这个Redisson也都替我们考虑到了，通过Redisson在github上的[官方文档][Link 1]就可以了解（官方文档甚至还有中文版，简直良心）  
![多种不同的锁][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xPX1lVTg_size_16_color_FFFFFF_t_70_pic_center]  
Redisson支持以上这么多种复杂的锁的语义，我这里就不一个个来分析了，大致的思路和最基础的可重入锁是一致的，使用不同的lua脚本即可实现，同时搭配上watchdog机制，就可以让我们轻松玩转各种不同场景的分布式锁

[Link 1]: https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xPX1lVTg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221122/b03c3817eef54e0b8ac11b59dad643ba.png