🔥博客主页： 【小扳_-CSDN博客】
❤感谢大家点赞👍收藏⭐评论✍

本章目录

        1.0 基于 Redis 实现的分布式锁存在的问题

        2.0 Redisson 功能概述

        3.0 Redisson 具体使用

        4.0 Redisson 可重入锁原理

        5.0 Redisson 锁重试原理

        6.0 Redisson WatchDog 机制

        6.1 Redisson 是如何解决超时释放问题的呢？

        7.0 Redisson MultiLock 原理

        7.1 Redisson 分布式锁是如何解决主从一致性问题的呢？

---

        1.0 基于 Redis 实现的分布式锁存在的问题

        首先，在之前基于 setnx 实现的分布式锁存在以下问题：

        1）不可重入：同一个线程无法多次获取同一把锁。

        2）不可重试：获取锁只尝试一次就返回 false ，没有重试机制。

        当然这个机制是可以自己在判断完有无获取锁之后，再来根据业务的需求进行手动添加代码。比如说，当业务需求是：需要重复尝试获取锁。则可以在判断获取锁失败之后，等待一段时间，再去获取锁即可。

        3）超时释放：锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患。

        比如说，当业务阻塞时间较久，锁到了超时时间则会自动释放，那么其他线程就会有可能获取锁成功，这就出现了多个线程获取锁成功，从而导致线程安全问题。

        4）主从一致性：如果 Redis 提供了主从集群，主从同步延迟，当主机宕机时，如果未来得及同步到其他机器上，则就会出现多线程获取锁成功情况，从而导致线程安全问题。

        那么 Java 实现了解决以上问题的 Redisson 分布式服务类。

        2.0 Redisson 功能概述

        Redisson 是一个在 Redis 的基础上实现的 Java 驻内存数据网络。它不仅提供了一系列的分布式的 Java 常用对象，还提供了许多分布式服务，其中包含了各种分布式锁的实现。

        Redisson 解决了不可重入问题、不可重试问题、超时释放问题、主从一致性问题。

        比如说，分布式锁的可重入锁、公平锁、联锁、红锁等等。

        3.0 Redisson 具体使用

        1）引入依赖

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

        2）配置 RedissonClient类

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient client(){
        //配置类
        Config config = new Config();
        //添加redis地址，这里添加了单点的地址，也可以使用config.useClusterServers()添加集群地址
        config.useSingleServer().setAddress("redis://8.152.162.159:6379").setPassword("****");
        //创建客户端
        return Redisson.create(config);
    }
}

        3）使用 RedissonClient类

    @Autowired
    RedissonClient redissonClient;

    @Test
    void contextLoads() throws InterruptedException {
        //先获取锁对象，根据业务来锁定资源
        RLock lock = redissonClient.getLock("lock");

        //尝试获取锁
        //tryLock() 进行了重写，有无参、只有两个参数、有三个参数
        boolean b = lock.tryLock(1, TimeUnit.SECONDS);
        
        if (b){
            System.out.println("成功获取锁!");
        }else {
            System.out.println("获取锁失败!");
        }
        
    }

        先注入 RedissonClient 对象，根据 getLock("锁") 方法获取 RLock lock 锁对象，根据业务需要对资源进行锁定。  

        调用 lock 对象中的 tryLock() 方法来尝试获取锁，该方法进行了重写：

        1）boolean tryLock()：当获取锁失败时，默认不等待，就是不重试获取锁，默认锁的超时时间为 30 秒。

        2）boolean tryLock(long time, TimeUnit unit)：在 time 时间内会进行重试尝试获取锁，unit 为时间单位。默认锁的超时时间为 30 秒。

        3）boolean tryLock(long waitTime, long leaseTime, TimeUnit unit)：在获取锁失败时，在 waitTime 时间内进行重试尝试获取锁，锁的超时时间为 leaseTime 秒，unit 为时间单位。

        最后，调用 lock 对象中的方法 unlock() 来释放锁。

具体代码：

    @Autowired
    RedissonClient redissonClient;

    @Test
    void contextLoads() throws InterruptedException {
        //先获取锁对象，根据业务来锁定资源
        RLock lock = redissonClient.getLock("lock");

        //尝试获取锁
        //tryLock() 进行了重写，有无参、只有两个参数、有三个参数
        boolean b = lock.tryLock(1, TimeUnit.SECONDS);

        if (!b){
            System.out.println("获取锁失败!");
        }

        try {
            System.out.println("获取锁成功!");
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            //释放锁
            lock.unlock();
        }

    }

        4.0 Redisson 可重入锁原理

        在之前的基于 setnx 实现的分布式锁是不支持可重入锁，举个例子：线程一来获取锁，使用 setnx 来设置，当设置成功，则获取锁成功了，线程一在获取锁成功之后，再想来获取相同的锁时，则再次执行 setnx 命令，那一定是不可能成功获取，因为 setxn 已经存在了，这就是基于 setnx 来实现分布式锁不可重入锁的核心原因。

        而对于 Redisson 可以实现可重入锁，这是如何实现的呢？

        其核心原因是基于 Redis 中的哈希结构实现的分布式锁，利用 key 来锁定资源，对于 field 来标识唯一成功获取锁的对象，而对于 value 来累计同一个线程成功获取相同的锁的次数。

        具体实现思路：

        1）尝试获取锁：

        先判断缓存中是否存在 key 字段，如果存在，则说明锁已经被成功获取，这时候需要继续判断成功获取锁的对象是否为当前线程，如果根据 key field 来判断是当前线程，则 value += 1 且还需要重置锁的超时时间；如果根据 key field 判断不是当前线程，则直接返回 null。如果缓存中不存在 key 字段，则说明锁还没有被其他线程获取，则获取锁成功。

        2）释放锁：

        当业务完成之后，在释放锁之前，先判断获取锁的对象是不是当前线程，如果不是当前线程，则说明可能由于超时，锁已经被自动释放了，这时候直接返回 null；如果是当前线程，则进行 value -= 1 ，最后再来判断 value 是否大于 0 ，当大于 0 时，则不能直接释放锁，需要重置锁的超时时间；当 value = 0 时，则可以真正的释放锁。

如图：

 

        又因为使用 Java 实现不能保证原子性，所以需要借助 Lua 脚本实现多条 Redis 命令来保证原则性。

尝试获取锁的 Lua 脚本：

释放锁的 Lua 脚本：

        5.0 Redisson 锁重试原理

        在之前基于 setnx 实现的分布式锁，获取锁只尝试一次就返回 false ，没有重试机制。

        而 Redisson 是如何实现锁重试的呢？

实现锁重试

        追踪源代码：

得到该类：

        首先，将等待时间转换为毫秒，接着获取当前时间和获取当前线程 ID ，再接着第一个尝试去获取锁，将参数 waitTime 最大等待时间，leaseTime 锁的超时时间，unit 时间单位，threadId 当前线程 ID 传进去 tryAcquire 方法中。

        紧接着来查看 tryAcquire 方法：

         再查看调用的 tryAcquireAsync 方法：

        当指定了 leaseTime 锁的超时时间，则会调用 tryLockInnerAsync 方法；当没有指定 leaseTime 锁的超时时间，则会调用 getLockWatchdogTimeout 方法，默认超时时间为 30 秒。

        接着查看 tryLockInnerAsync 方法：

         可以看到，这就是尝试获取是的 Lua 脚本执行多条 Redis 命令。

        细心可以发现，如果正常获取锁，则返回 null ；如果获取锁失败，则返回当前锁的 TTL ，锁的剩余时间。

        因此最后将当前锁的 TTL 返回赋值给 Long ttl 变量。

        再接着往下：

        当 ttl == null ，则说明当前线程成功获取锁，因此就不需要接着往下再次尝试去获取锁了。相反，当 ttl != null ，则需要接着往下走，重新尝试去获取锁。

        判断 time 等于当前时间减去在第一次获取锁之前的时间，time 也就是最大的等待时间还剩多少。判断 time 是否小于 0 ，若小于 0 则已经到了最大等待时间了，所以不需要再继续等下去了，直接返回 false 即可。

        若 time 还是大于 0 ，则接着往下走：

        调用 subscribe 方法，该方法可以理解成订阅锁，一旦锁被释放之后，该方法就会收到通知，然后再去尝试获取锁。

回顾在释放锁的时候，使用 Redis 命令中的 redis.call('publish', KEYS[2], ARGV[1]) 来发布消息，通知锁已经被释放，一旦锁被释放，那么就可以成功订阅。

        因此，在订阅锁的过程中，并不是一直死等下去，而是在 time 剩余最大等待时间之内，如果可以订阅锁成功，才会去尝试获取锁。如果在 time 时间内，订阅锁失败，则会取消订阅，再返回 false 。

        接着往下走，当在 time 时间内订阅锁成功，会更新 time 时间，也就是更新最大的等待时间，判断 time 小于 0 ，则返回 false ，如果 time 还是大于 0 ，则到了真正尝试第二次获取锁，调用 tryAcquire(waitTime, leaseTime, unit, threadId) 方法，将返回值再次赋值给变量 ttl ，判断 ttl == null ，则说明成功获取锁了，直接返回 true ；判断 ttl != null ，则第二次获取锁还是失败，由需要更新 time 了，因为在调用尝试获取锁的过程中，消耗时间还是挺大的，同理，判断更新完之后的 time 是否大于 0，如果 time 小于 0，则超过了剩余最大锁的超时时间，返回 false ；

        如果判断 time 仍旧大于 0 ：

        那么先判断锁的过期时间 ttl 与 剩余时间 time ，如果 ttl < time ，则类似订阅方法一样的思路，选择等待 ttl 锁的过期时间，当 ttl 过期之后，就会订阅该锁；如果 time < ttl ，则 ttl 还没有释放，就不需要等 ttl 了，等到 time 结束还没有订阅到锁，则 time 也就小于 0 了，如果在 time 时间内获取到锁，再次尝试去获取锁，同样的，当在 ttl 时间内，成功订阅了，而且 time > 0 ，则会第三次去尝试获取锁。之后的步骤都是如此，这里使用了 do whlie 循环，判断循环成立为 time > 0，当 time < 0 ，则会退出循环。

        总结，在解决可重试锁过程中，并不是循环不断的调用 tryAcquire(waitTime, leaseTime, unit, threadId) 方法来获取锁，这样容易造成 CPU 的浪费，而是通过等待锁释放，再去获取锁的方式来实现的可重试锁，利用信号量（Semaphore）和发布/订阅（PubSub）模式实现等待、唤醒、获取锁失败的重试机制。

        6.0 Redisson WatchDog 机制

        在之前基于 setnx 实现的分布式锁，锁超时释放虽然可以避免死锁，但是如果是业务执行耗时较长，也会导致锁释放，存在安全隐患。

        6.1 Redisson 是如何解决超时释放问题的呢？

        解决超时释放的核心是：当 leaseTime == -1 时，为了保证当前业务执行完毕才能释放锁，而不是业务还没有执行完毕，锁就被自动释放了。

追踪源代码：

        当 leaseTime == -1 时，默认锁的最大超时时间为 30 秒，会执行以下代码。

        接着点进去：

        WatchDog 会在锁的过期时间到期之前，定期向 Redis 发送续约请求，更新锁的过期时间。这通常是通过设置一个较短的过期时间和一个续约间隔来实现的。

        如果持有锁的线程正常释放锁，WatchDog 会停止续约操作。如果持有锁的线程崩溃或失去响应，WatchDog 会在锁的过期时间到达后自动释放锁。

        简单概述一下 WatchDog 机制：在获取锁成功之后，就会调用 scheduleExpirationRenewal(threadId) 方法开启自动续约，具体是由在 map 中添加业务名称和任务定时器，这个定时器会在一定时间内执行，比如说 10 秒就会自动开启任务，而该定时器中的任务就是不断的重置锁的最大超时时间，使用递归，不断的调用重置锁的时间，这就保证了锁是永久被当前线程持有。 

        这样就可以保证执行业务之后，才会释放锁。释放锁之后，会取消定时任务。

        7.0 Redisson MultiLock 原理

        7.1 Redisson 分布式锁是如何解决主从一致性问题的呢？

        先搞清楚什么是主从一致性问题，在集群的 Redis 中会区分出主力机和一般机器，在写 Redis 命令会放到主力机中运行，而主力机和一般机器需要保证数据都是一样的，也就是主从同步数据，在主力机中执行写命令时，突然发生宕机，未来得及将数据同步到其他一般机器中，而且当主力机宕机之后，会选出一台一般机器充当主力机，这时候的主力机没有同步之前的数据，那么其他线程再来写命名的时候就会出现问题了，这出现了主从不一致性。

        那么 Redisson 是如何来解决该问题呢？

        在多主架构中，每台主机都可以接收写请求，这样即使某一台主机宕机，其他主机仍然可以继续处理写请求。

        当某一台主机宕机后，如果在它恢复之前有新的写操作发生，可能会导致数据不一致。通过比较不同主机的数据状态，可以很容易地发现这些不一致的问题。

        当宕机的主机恢复后，可以通过与其他主机的数据进行比较，找出差异并进行数据同步，确保所有主机的数据一致。

        简单来说，设置多台主力机，每一次写命令都是一式多份，当某一台主力机出现宕机了，主从未来得及同步时，再写命令，同样一式多份，这样充当主力机出现了跟其他主力机不同的结果时，就很容易的发现问题了。

        通过设置多台主力机并进行写操作的多份复制，可以有效提高系统的可靠性，并在出现问题时快速发现和解决数据不一致的问题。

具体使用：