图解Redisson如何实现分布式锁、锁续约？

快来打我* 2023-09-24 15:06 89阅读 0赞

#### 文章目录 ####

*  一、基础
 *   *  0）Redisson版本说明、案例
     *   *  案例
     *  1）Redisson连接Redis的方式
     *  2）用到的Redis命令
     *  3）用到的lua脚本语义
 *  二、源码分析
 *   *  1、RLock
     *   *  获取RLock对象
     *  2、加锁流程
     *   *  0）加锁流程图
         *  1）加锁到哪台机器
         *  2）Client第一次加锁
         *  3）加锁成功之后的锁续约
         *  4）重入加锁（相同线程多次加锁）
         *  5）锁竞争（其他线程加锁失败）
         *   *  1> 一直重试直到加锁成功
             *  2> 等待锁超时返回加锁失败
     *  3、释放锁流程
     *   *  1）Client主动尝试释放锁
         *  2）Client主动强制释放锁
         *  3）Client宕机，锁超时释放
         *  4）不启动锁续约的超时释放锁

## 一、基础 ##

### 0）Redisson版本说明、案例 ###

使用当前（2022年12月初）最新的版本：3.18.1；

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.18.1</version>
    </dependency>

#### 案例 ####

案例采用redis-cluster集群的方式；

public class Main {
        
        public static void main(String[] args) throws Exception {
        
            // 1.配置Redis-Cluster集群节点的ip和port 
            Config config = new Config();
            config.useClusterServers()
                    .addNodeAddress("redis://127.0.0.1:7001")
                    .addNodeAddress("redis://127.0.0.1:7002")
                    .addNodeAddress("redis://127.0.0.1:7003")
                    .addNodeAddress("redis://127.0.0.1:7004");
            // 2.创建Redisson的客户端 
            RedissonClient redisson = Redisson.create(config);
            // 3.测试Redisson可重⼊锁的加锁、释放锁
            testLock(redisson);
        }
    
        private static void testLock(RedissonClient redisson) throws InterruptedException {
        
            // 1.获取key为"anyLock"的锁对象
            final RLock lock = redisson.getLock("test_lock");
            boolean locked = true;
            try {
        
                //2.1：加锁 
                lock.lock();
                // 2.2：加锁，并设置尝试获取锁超时时间30s、锁超时⾃动释放的时间10s 
    //            locked = lock.tryLock(30, 10, TimeUnit.SECONDS);
                if (locked)
                    System.out.println("加锁成功！" + new Date());
                
                Thread.sleep(20 * 1000);
                System.out.println("锁逻辑执行完毕！" + new Date());
    
            } finally {
        
                // 3.释放锁 
                lock.unlock();
            }
        }
    }

### 1）Redisson连接Redis的方式 ###

redission支持4种连接redis方式，分别为单机、主从、Sentinel、Cluster 集群；在分布式锁的实现上区别在于hash槽的获取方式。

具体配置方式见Redisson的GitHub（[https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95\#21-%E7%A8%8B%E5%BA%8F%E5%8C%96%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95][https_github.com_redisson_redisson_wiki_2.-_E9_85_8D_E7_BD_AE_E6_96_B9_E6_B3_95_21-_E7_A8_8B_E5_BA_8F_E5_8C_96_E9_85_8D_E7_BD_AE_E6_96_B9_E6_B3_95]）

![在这里插入图片描述][5df57d369956477d9f4cb00b4898e6f0.png]

### 2）用到的Redis命令 ###

**分布式锁主要需要以下redis命令：**

> 1.  **EXISTS key**：当 key 存在，返回1；不存在，返回0。
> 2.  **GETSET key value**：将给定 key 的值设为 value ，并返回 key 的旧值 (old value)；当 key 存在但不是字符串类型时，返回一个错误；当key不存在时，返回nil。
> 3.  **GET key**：返回 key 所关联的字符串值，如果 key 不存在那么返回 nil。
> 4.  **DEL key \[KEY …\]**：删除给定的一个或多个 key（不存在的 key 会被忽略），返回实际删除的key的个数（integer）。
>     
>         DEL key1 key2 key3
> 5.  **HSET key field value**：给一个key 设置一个\{field=value\}的组合值，如果key没有就直接赋值并返回1；如果field已有，那么就更新value的值，并返回0。
> 6.  **HEXISTS key field**：当key中存储着field的时候返回1，如果key或者field有一个不存在返回0。
> 7.  **HINCRBY key field increment**：将存储在key中的哈希（Hash）对象中的指定字段field的值加上增量increment；
>     
>      *  如果键key不存在，一个保存了哈希对象\{field=value\}的key将被创建；
>      *  如果字段field不存在，在进行当前操作前，feild将被创建，且对应的值被置为0；返回值是increment。
> 8.  **PEXPIRE key milliseconds**：设置存活时间，单位是毫秒。**EXPIRE**操作单位是秒。
> 9.  **PUBLISH channel message**：向channel post一个message内容的消息，返回接收消息的客户端数。

### 3）用到的lua脚本语义 ###

**Redisson源码中，执行redis命令的是lua脚本，其中主要有如下几个概念：**

> 1.  **redis.call()**：执行redis命令。
> 2.  KEYS\[n\]：指脚本中第n个参数，比如KEYS\[1\]指脚本中的第一个参数。
> 3.  ARGV\[n\]：指脚本中第n个参数的值，比如ARGV\[1\]指脚本中的第一个参数的值。
> 4.  返回值中nil与false同一个意思。

在redis执行lua脚本时，相当于一个redis级别的锁，不能执行其他操作，类似于原子操作，这也是redisson实现的一个关键点。

另外，如果lua脚本执行过程中出现了异常或者redis服务器宕机了，会将脚本中已经执行的命令在AOF、RDB日志中删除；即LUA脚本执行报错会进行回滚操作。

## 二、源码分析 ##

### 1、RLock ###

![在这里插入图片描述][56cb898f6091462c80c9ee6407e88f06.png]

RLock接口主要继承了Lock接口，并扩展了部分方法，比如：tryLock(long waitTime, long leaseTime, TimeUnit unit)方法中加入的leaseTime参数，用来设置锁的过期时间，如果超过leaseTime还没有解锁的话，redis就强制解锁；leaseTime的默认时间是30s。

#### 获取RLock对象 ####

RLock lock = redissonClient.getLock("test_lock");

RLock对象表示⼀个锁对象，我们要某一个key加锁时，需要先获取⼀个锁对象。

![在这里插入图片描述][be906f8d0e1f4df6a39596ed24576923.png]

这里并没有具体请求Redis进行加锁的逻辑，而只是调用RedissonLock的构造函数，设置一些变量。

### 2、加锁流程 ###

进入到Rlock\#lock()方法，先看主流程；关于竞争锁等待时间、锁超时释放时间的配置、使用，在流程中穿插着聊。

#### 0）加锁流程图 ####

![请添加图片描述][4deef5c571994e6b953a15b325d5461d.png]

#### 1）加锁到哪台机器 ####

**lock()方法执行链路：**

![在这里插入图片描述][46fd69c8ad2c4e94a2f181e215276727.png]

走到这里，已经可以看到加锁的底层逻辑：LUA脚本。

**而lua脚本只是⼀⼤串字符串，作为evalWriteAsync()⽅法的⼀个参数⽽已；所以下⼀步进到evalWriteAsync()⽅法中：**

![在这里插入图片描述][9432c5f1b92b42cdb67eac802ee1bf6e.png]

走到这里会调用ConnectionManager\#getEntry(String)方法；

在创建RedissonClient时，笔者配置的是Redis-Cluster，而走到这里却会进入到`MasterSlaveConnectionManager`，实际上实例化的ConnectionManager就是RedisCluster模式下的ClusterConnectionManager，而`ClusterConnectionManager`继承自`MasterSlaveConnectionManager`，并且`ClusterConnectionManager`没有重写getEntry(String)方法，所以会进入到`MasterSlaveConnectionManager`\#getEntry(String)方法。

ConnectionManager\#getEntry(String)方法会根据传入的key名称找到相应的Redis节点、目标master。

Redis-Cluster集群中的数据分布式是 通过⼀个⼀个的hash slot来实现的，Redis-Cluster集群总共16384个hash slot，它们都 会被均匀分布到所有的master节点上；**这里ClusterConnectionManager通过key名称计算出相应的hash slot方式如下：**

*  ⾸先通过key计算出CRC16值，然后 CRC16值对16384进⾏取模，进⽽得到hash slot。

@Override
    public int calcSlot(String key) {
        
        if (key == null) {
        
            return 0;
        }
    
        int start = key.indexOf('{');
        if (start != -1) {
        
            int end = key.indexOf('}');
            if (end != -1 && start + 1 < end) {
        
                key = key.substring(start + 1, end);
            }
        }
    
        int result = CRC16.crc16(key.getBytes()) % MAX_SLOT;
        log.debug("slot {} for {}", result, key);
        return result;
    }

这⾥计算出key的hash slot之后，就可以通过hash slot 去看⼀看哪个master上有这个hash slot，如果某个master上有个这个hash slot，那么这个 key当然就会落到该master节点上，执⾏加锁指令也就应该在该master上执⾏。

**下面进入本文重点，可重入锁的各种加锁、释放锁。**

#### 2）Client第一次加锁 ####

在寻找应该在哪台Redis机器上加锁时，在RedissonLock\#tryLockInnerAsync()方法中我们看到了一堆LUA脚本：

![在这里插入图片描述][5a9359e9b5344952807cdb3206d98ce3.png]

**LUA脚本参数解析：**

>  *  KEYS\[1\] 表示的是 getName() ，即锁key的名称，比如案例中的 test\_lock；
>  *  ARGV\[1\] 表示的是 internalLockLeaseTime 默认值是30s；
>  *  ARGV\[2\] 表示的是 getLockName(threadId) ，唯一标识当前访问线程，使用锁对象id+线程id（UUID:ThreadId）方式表示，用于区分不同服务器上的线程。
>     
>      *  UUID用来唯⼀标识⼀个客户端，因为会有多个客户端的多个线程加锁；
>      *  结合起来的`UUID:ThreadId` 表示：具体哪个客户端上的哪个线程过来加锁，通 过这样的组合⽅式唯⼀标识⼀个线程。

**LUA脚本逻辑：**

> 1.  如果锁名称不存在；
>     
>      *  则向redis中添加一个key为test\_lock的HASH结构、添加一个field为线程id，值=1的键值对\{field:increment\}，表示此线程的重入次数为1；
>      *  设置test\_lock的过期时间，防止当前服务器出问题后导致死锁，然后`return nil; end;`返回nil，lua脚本执行完毕；
> 2.  如果锁存在，检测当前线程是否持有锁；
>     
>      *  如果是当前线程持有锁，`hincrby`将该线程重入的次数++；并重新设置锁的过期时间；返回nil，lua脚本执行完毕；
>      *  如果不是当前线程持有锁，`pttl`返回锁的过期时间，单位ms。

第一次加锁时，key肯定不存在与master节点上；会执行下列LUA脚本对应的Redis指令：

hset test_lock UUID:ThreadId 1 
    pexpire test_lock 30000

此时，Redis中多一个Hash结构的key（test\_lock）：

test_lock : 
    {
        
        UUID:ThreadId：1
    }

这里的1使用来做锁重入的。

`pexpire`指令为test\_lock这个key设置过期时间为30s，即：30s后这个key会⾃动过期被删除，key对应的锁在那时也就被释放了。

**总体来看，加锁的逻辑很简单：**

*  在key对应的hash数据结构中记录了⼀ 下当前是哪个客户端的哪个线程过来加锁了，然后设置了⼀下key的过期时间为30s。

#### 3）加锁成功之后的锁续约 ####

成功加锁后，lua脚本返回nil，即null。

![在这里插入图片描述][3edbd13fc88648689e20ca1ea50a17b6.png]

加锁成功之后，tryLockInnerAsync()⽅法返回；再结合Java8的Stream，对加锁结果进一步处理；

因为加锁成功后返回的是nil，这是lua脚本的返回形式，体现到java代码中的返回值为：null。  
又由于RLock\#lock()方法传入的leaseTime是-1，所以进入到`scheduleExpirationRenewal(long)`方法做锁续约。

![在这里插入图片描述][17171c088d444006b17734c678c56c07.png]

**renewExpirationAsync()方法负责做具体的锁续约：**

protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
        
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0;",
                Collections.singletonList(getRawName()),
                internalLockLeaseTime, getLockName(threadId));
    }

**这里LUA脚本的逻辑很简单：**

*  判断当前key中，是否还被线程UUID:ThreadId持有锁，持有则设置过期时间为30s（续命）。

锁续约（看门狗机制）其实就是每次加锁成功后，会⻢上开启⼀个后台线程， 每隔10s检查⼀下key是否存在，如果存在就为key续期30s。

*  这里的10s，取自配置的`lockWatchdogTimeout`参数，默认为30 \* 1000 ms；
 *  所以⼀个key往往当过期时间慢慢消逝到20s左右时就⼜会被定时任务重置为了30s，这样就能保证：只要这个定时任务还在、这个key还在，就⼀直维持加锁。

**如果当前持有锁的线程被中断了，会停止锁续约，即杀死看门狗；**

![在这里插入图片描述][22114b25a7f9446697373acf31fb9942.png]

protected void cancelExpirationRenewal(Long threadId) {
        
        ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (task == null) {
        
            return;
        }
        
        if (threadId != null) {
        
            task.removeThreadId(threadId);
        }
    
        if (threadId == null || task.hasNoThreads()) {
        
            Timeout timeout = task.getTimeout();
            if (timeout != null) {
        
                timeout.cancel();
            }
            EXPIRATION_RENEWAL_MAP.remove(getEntryName());
        }
    }

所谓的停止锁续约，实际就是将当前线程的threadId从看门狗缓存中移除，后续在执行锁续约时，如果发现看门狗缓存中已经没有了当前线程threadId，则直接退出锁续约 并且 不再延时10s开启一个定时任务。

**如果加锁时指定了leaseTime > 0，则不会开门狗机制，表示强制锁leaseTime 毫秒后过期。一共有三种加锁方式可以做到，如下：**

>  *  RLock\#lock(long leaseTime, TimeUnit unit)
>  *  RLock\#tryLock(long waitTime, long leaseTime, TimeUnit unit)
>  *  RLock\#lockInterruptibly(long leaseTime, TimeUnit unit)

#### 4）重入加锁（相同线程多次加锁） ####

再次回到加锁的LUA脚本：

![在这里插入图片描述][e992a38c6e4a4112ae70d1555adc47db.png]

同一个线程对分布式锁多次加锁时，会走以下逻辑：

*  判断当前key是否被当前线程持有，如果是则增加锁重入的次数，并重新设置锁的过期时间为30s；

**对应的Redis命令为：**

hexists test_lock UUID:ThreadId
    hincrby test_lock UUID:ThreadId 1
    pexpire test_lock 30000

此时Redis中test\_key对应的数据结构从

test_lock : 
    {
        
        UUID:ThreadId：1
    }

**变成：**

test_lock : 
    {
        
        UUID:ThreadId：2
    }

并将key的过期时间重新设置为30s。

锁重入成功之后，后台也会开启⼀个watchdog后台线程做锁续约，每隔10s检查⼀下key，如果key存在就将key的过期时间重新设置为30s。

Redisson可重⼊加锁的语义，实际是通过Hash结构的key中某个线程（UUID:ThreadId）对应的加锁次数来表示的。

#### 5）锁竞争（其他线程加锁失败） ####

再再次回到加锁的LUA脚本：

![在这里插入图片描述][9ad82ca0576949eab10561716cd3305f.png]

**如果分布式锁已经被其他线程持有，LUA脚本会执行以下逻辑：**

*  返回当前key的剩余存活时间，因为不是返回nil，也就代表着加锁失败；

**对应的Redis的命令为：**

pttl test_lock

针对加锁方式的不同，加锁失败的逻辑也不同；可以分两大类：指定了加锁失败的等待时间waitTime和未指定waitTime。

*  未执行加锁失败的等待时间`waitTime`：获取分布式锁失败会一直重试，直到获取锁成功。比如下列加锁方法：
    
     *  `Rlock#lock()`：一直尝试获取分布式锁，直到获取锁成功。
     *  `RLock#lockInterruptibly(long leaseTime, TimeUnit unit)`
     *  `RLock#lock(long leaseTime, TimeUnit unit)`
 *  指定了加锁失败的等待时间`waitTime`：获取分布式锁会超时，超时之后返回加锁失败；
    
     *  `Rlock#tryLock(long waitTime, TimeUnit unit)`：指定获取锁失败的等待时间。在等待时间范围之内进行重试，超时则返回加锁失败。
     *  `Rlock#tryLock(long waitTime, long leaseTime, TimeUnit unit)`：同样是指定获取锁失败的等待时间，并且强制指定锁过期的时间（不开启看门狗）。在等待时间范围之内进行重试，超时则返回加锁失败。

**可以简单的概述为RLock接口下的`tryLock()`方法获取锁会失败，`lock()`方法获取锁一定会成功。**

##### 1> 一直重试直到加锁成功 #####

以`Rlock#lock()`方法为例：

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
        
            return;
        }
    
        CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
        pubSub.timeout(future);
        RedissonLockEntry entry;
        if (interruptibly) {
        
            entry = commandExecutor.getInterrupted(future);
        } else {
        
            entry = commandExecutor.get(future);
        }
    
        try {
        
            while (true) {
        
                // lock() 或 lockInterruptibly()为入口走到这里时。leaseTime为-1，表示会开始开门狗；如果leaseTime大于0，则不会开启开门狗；
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
        
                    break;
                }
    
                // waiting for message
                if (ttl >= 0) {
        
                    try {
        
                        // 因为Semaphore的可用资源为0，所以这里就等价于Thread.sleep(ttl)；
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
        
                        if (interruptibly) {
        
                            throw e;
                        }
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
        
                    if (interruptibly) {
        
                        entry.getLatch().acquire();
                    } else {
        
                        entry.getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
        
            unsubscribe(entry, threadId);
        }
    }

> 1.  首先订阅解锁channel（命名格式：`redisson_lock__channel:{keyName}`），其他线程解锁后，会发布解锁的消息；这里收到消息会立即尝试获取锁；订阅解锁channel的超时时间默认为7.5s。也就说获取锁失败7.5s之内，如果其他线程释放锁，当前线程可以立即尝试获取到锁。
> 2.  获取锁失败之后会进⼊⼀个while死循环中：
>     
>      *  每休息锁的存活时间`ttl`之后，就尝试去获取锁，直到成功获取到锁才会跳出while死循环。

##### 2> 等待锁超时返回加锁失败 #####

以`Rlock#tryLock(long waitTime, TimeUnit unit)`为例：

@Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
        
            return true;
        }
    
        // 获取锁剩余的等待时长
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
        
            // 获取锁超时，返回获取分布式锁失败
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
        CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        try {
        
            // 订阅解锁channel的超时时长为 获取锁剩余的等待时长
            subscribeFuture.get(time, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
        
            if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(
                    "Unable to acquire subscription lock after " + time + "ms. " +
                            "Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {
        
                subscribeFuture.whenComplete((res, ex) -> {
        
                    if (ex == null) {
        
                        unsubscribe(res, threadId);
                    }
                });
            }
            acquireFailed(waitTime, unit, threadId);
            return false;
        } catch (ExecutionException e) {
        
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
    
        try {
        
            // 收到解锁channel的消息之后，走到这里，再次判断获取锁等待时长是否超时
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
        
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        
            // while循环中尝试去获取锁
            while (true) {
        
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
        
                    return true;
                }
    
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
        
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }
    
                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
        
                    // 如果获取锁失败后，锁存活时长 小于 剩余锁等待时长，则线程睡眠 锁存活时长
                    commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
        
                    // 如果获取锁失败后，锁存活时间 大于等于 剩余锁等待时长，则线程睡眠 锁等待时长
                    commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }
    
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
        
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }
            }
        } finally {
        
            unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
        }
    }

加锁存在超时时间 相比于 一直重试直到加锁成功，只是多一个时间限制，具体差异体现在：订阅解锁channel的超时时长、获取锁失败后线程的睡眠时长、重试获取锁次数的限制；

> 1.  获取分布式锁失败之后，立即判断当前获取锁是否超时，如果超时，则返回加锁失败；
> 2.  否者，订阅解锁channel（命名格式：`redisson_lock__channel:{keyName}`），其他线程解锁后，会发布解锁的消息；订阅解锁channel的超时时间为 `获取锁剩余的等待时长`。
>     
>      *  在这个时间范围之内，如果其他线程释放锁，当前线程收到解锁channel的消息之后再次判断获取锁是否超时，如果不超时，尝试获取锁。
> 3.  获取锁之后会进⼊⼀个while死循环中：
>     
>      *  如果获取锁超时，则返回加锁失败；
>      *  否者让线程睡眠：
>         
>          *  如果锁存活时长`ttl` 小于 剩余锁等待时长，则线程睡眠 锁存活时长；
>          *  如果锁存活时间`ttl` 大于等于 剩余锁等待时长，则线程睡眠 锁等待时长；
>      *  线程睡眠完之后，判断获取锁是否超时，不超时则尝试去获取锁。

### 3、释放锁流程 ###

#### 1）Client主动尝试释放锁 ####

进入到Rlock\#unlock()方法；

![在这里插入图片描述][3f09240e9fcf4ad29d0702a3bb78ab5d.png]

和加锁的方式⼀样，释放锁也是通过lua脚本来完成的；

**LUA脚本参数解析：**

>  *  KEYS\[1\] 表示的是 getName() ，代表的是锁名 test\_lock；
>  *  KEYS\[2\] 表示getChanelName() 表示的是发布订阅过程中使用的Chanel；
>  *  ARGV\[1\] 表示的是LockPubSub.unLockMessage，解锁消息，实际代表的是数字 0，代表解锁消息；
>  *  ARGV\[2\] 表示的是internalLockLeaseTime 默认的有效时间 30s；
>  *  ARGV\[3\] 表示的是 getLockName(thread.currentThread().getId()) 代表的是 UUID:ThreadId 用锁对象id+线程id， 表示当前访问线程，用于区分不同服务器上的线程。

**LUA脚本逻辑：**

> 1.  如果锁名称不存在；
>     
>      *  可能是因为锁过期导致锁不存在，也可能是并发解锁。
>      *  则发布锁解除的消息，返回1，lua脚本执行完毕；
> 2.  如果锁存在，检测当前线程是否持有锁；
>     
>      *  如果是当前线程持有锁，定义变量counter，接收执行`incrby`将该线程重入的次数–的结果；
>         
>          *  如果重入次数大于0，表示该线程还有其他任务需要执行；重新设置锁的过期时间；返回0，lua脚本执行完毕；
>          *  否则表示该线程执行结束，`del`删除该锁；并且`publish`发布该锁解除的消息；返回1，lua脚本执行完毕；
>      *  如果不是当前线程持有锁 或 其他情况，都返回nil，lua脚本执行完毕。

脚本执行结束之后，如果返回值不是0或1，即当前线程去释放其他线程的加锁时，抛出异常。

**通过LUA脚本释放锁成功之后，会将看门狗杀死；**

![在这里插入图片描述][7a5fef642c0c400d92076dd6f96c72b4.png]

#### 2）Client主动强制释放锁 ####

`forceUnlockAsync()`方法被调用的地方很多，大多都是在清理资源时删除锁。

@Override
    public RFuture<Boolean> forceUnlockAsync() {
        
        cancelExpirationRenewal(null);
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('del', KEYS[1]) == 1) then "
                        + "redis.call('publish', KEYS[2], ARGV[1]); "
                        + "return 1 "
                        + "else "
                        + "return 0 "
                        + "end",
                Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE);
    }

**LUA脚本逻辑：**

*  逻辑比较简单粗暴：删除锁成功则并发布锁被删除的消息，返回1结束，否则返回0结束。

#### 3）Client宕机，锁超时释放 ####

如果Redisson客户端刚加锁成功，并且未指定releaseTime，后台会启动一个定时任务watchdog每隔10s检查key：key如果存在就为它⾃动续命到30s；在watchdog定时任务存在的情况下，如果不是主动释放锁，那么key将会⼀直的被watchdog这个定时任务维持加锁。

但是如果客户端宕机了，定时任务watchdog也就没了，也就没有锁续约机制了，那么过完30s之后，key会⾃动被删除、key对应的锁也自动被释放了。

#### 4）不启动锁续约的超时释放锁 ####

如果在加锁时指定了leaseTime，加锁成功之后，后台并不会启动一个定时任务watchdog做锁续约；key存活leaseTime 毫秒之后便会自动被删除、key对应的锁也就自动被释放了；无论当前线程的业务逻辑是否执行完毕。

**比如使用如下方式加锁：**

*  RLock\#lock(long leaseTime, TimeUnit unit)
 *  RLock\#tryLock(long waitTime, long leaseTime, TimeUnit unit)
 *  RLock\#lockInterruptibly(long leaseTime, TimeUnit unit)

[https_github.com_redisson_redisson_wiki_2.-_E9_85_8D_E7_BD_AE_E6_96_B9_E6_B3_95_21-_E7_A8_8B_E5_BA_8F_E5_8C_96_E9_85_8D_E7_BD_AE_E6_96_B9_E6_B3_95]: https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95#21-%E7%A8%8B%E5%BA%8F%E5%8C%96%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95
[5df57d369956477d9f4cb00b4898e6f0.png]: https://img-blog.csdnimg.cn/5df57d369956477d9f4cb00b4898e6f0.png
[56cb898f6091462c80c9ee6407e88f06.png]: https://img-blog.csdnimg.cn/56cb898f6091462c80c9ee6407e88f06.png
[be906f8d0e1f4df6a39596ed24576923.png]: https://img-blog.csdnimg.cn/be906f8d0e1f4df6a39596ed24576923.png
[4deef5c571994e6b953a15b325d5461d.png]: https://img-blog.csdnimg.cn/4deef5c571994e6b953a15b325d5461d.png
[46fd69c8ad2c4e94a2f181e215276727.png]: https://img-blog.csdnimg.cn/46fd69c8ad2c4e94a2f181e215276727.png
[9432c5f1b92b42cdb67eac802ee1bf6e.png]: https://img-blog.csdnimg.cn/9432c5f1b92b42cdb67eac802ee1bf6e.png
[5a9359e9b5344952807cdb3206d98ce3.png]: https://img-blog.csdnimg.cn/5a9359e9b5344952807cdb3206d98ce3.png
[3edbd13fc88648689e20ca1ea50a17b6.png]: https://img-blog.csdnimg.cn/3edbd13fc88648689e20ca1ea50a17b6.png
[17171c088d444006b17734c678c56c07.png]: https://img-blog.csdnimg.cn/17171c088d444006b17734c678c56c07.png
[22114b25a7f9446697373acf31fb9942.png]: https://img-blog.csdnimg.cn/22114b25a7f9446697373acf31fb9942.png
[e992a38c6e4a4112ae70d1555adc47db.png]: https://img-blog.csdnimg.cn/e992a38c6e4a4112ae70d1555adc47db.png
[9ad82ca0576949eab10561716cd3305f.png]: https://img-blog.csdnimg.cn/9ad82ca0576949eab10561716cd3305f.png
[3f09240e9fcf4ad29d0702a3bb78ab5d.png]: https://img-blog.csdnimg.cn/3f09240e9fcf4ad29d0702a3bb78ab5d.png
[7a5fef642c0c400d92076dd6f96c72b4.png]: https://img-blog.csdnimg.cn/7a5fef642c0c400d92076dd6f96c72b4.png