【Redis】6、Redisson 分布式锁的简单使用（可重入、重试机制…）

目录

• 零、自己通过 set nx ex 实现的分布式锁存在的问题
• 一、Redisson 介绍
• 二、Redisson 基本使用（改造业务）
• • (1) 依赖
  • (2) 配置 Redisson 客户端
  • (3) 使用 Redisson 的可重入锁
• 三、Redisson 可重入锁原理
• 四、Redisson 可重试原理
• 五、Redisson 超时释放（锁的 ttl）
• 六、主从一致（连锁 MultiLock）
• 七、锁总结

零、自己通过 set nx ex 实现的分布式锁存在的问题

✏️ 不可重入 同一个线程无法多次获取同一把锁

✏️ 不可重试 获取锁只尝试一次就返回 false，没有重试机制

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

一、Redisson 介绍

✏️ Redisson 是一个在 Redis 基础上实现的 Java 驻内存数据网格（In-Memory Data Grid）

✏️ 它提供了一系列分布式的 Java 常用对象

✏️ 提供了许多分布式服务，其中包含了各种分布式锁的实现

官网地址 https://redisson.org

GitHub 地址 https://github.com/redisson/redisson

二、Redisson 基本使用（改造业务）

(1) 依赖

   
       org.redisson
       redisson
       3.13.6
   

(2) 配置 Redisson 客户端

@Configuration
@SuppressWarnings("all")
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.88.130:6379")
                .setPassword("root");
        return Redisson.create(config);
    }

}

✏️ 配置 redis 地址也可以使用 config.useClusterServers() 添加集群地址

✏️ useSingleServer() 是添加单点地址

(3) 使用 Redisson 的可重入锁

 @Test
 public void tesRedisson() throws InterruptedException {
     // 获取可重入锁, 指定锁的名称
     RLock lock = redissonClient.getLock("anLock");
     // 尝试获取锁
     // 参数1：获取锁的最大等待时间（期间会多次重试获取锁）
     // 参数2：锁自动释放时间
     // 参数3：时间单位
     boolean isGetLock = lock.tryLock(1, 10, TimeUnit.SECONDS);

     if (isGetLock) {
         try {
             System.out.println("执行业务");
         } finally {
             lock.unlock();
         }
     }
 }

三、Redisson 可重入锁原理

✏️ 通过 Redis 的 Hash 数据结构实现

✏️ 存入线程标识和 counter【记录锁被重入的次数（被使用的次数），被使用（获取）一次就递增1，被释放则 counter 减1】

✏️ 当 counter 为 0 的时候，整个锁会被释放

    @Resource
    private RedissonClient redissonClient;

    @SuppressWarnings("all")
    @Test
    public void tRedisson() {
        RLock lock = redissonClient.getLock("lock");
        boolean getLock = lock.tryLock();

        if (!getLock) {
            System.out.println("tRedisson getLock Failed");
            return;
        }

        try {
            System.out.println("tRedisson getLock Success");
            m(lock);
        } finally {
            System.out.println("tRedisson unlock");
            lock.unlock();
        }
    }

    public void m(RLock lock) {
        boolean getLock = lock.tryLock();

        if (!getLock) {
            System.out.println("m() getLock Failed");
            return;
        }

        try {
            System.out.println("m() getLock Success");
        } finally {
            System.out.println("m() unlock");
            lock.unlock();
        }
    }

✏️ 没有直接使用 hset

✏️ hincrby 命令：当 key 不存在的时候会自动创建 key

✏️ 获取锁成功：返回 nil；获取锁失败：返回当前锁的存活时间

✏️ pttl 以毫秒为单位返回某个 key 的存活时间

✏️ redis.call(‘publish’, KEYS[2], ARGV[1]); 发布释放锁的消息通知

四、Redisson 可重试原理

下面是 Redisson 的 tryLock() 方法的源码：

 @Override
 public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
     // 把超时等待时间转换为毫秒
     long time = unit.toMillis(waitTime);
     // 获取当前时间（毫秒）
     long current = System.currentTimeMillis();
     // 获取线程 ID
     long threadId = Thread.currentThread().getId();
     // tryAcquire 尝试获取锁（获取锁成功返回 null, 获取锁失败返回剩余ttl)
     Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
     // lock acquired
     if (ttl == null) { // 成功获取锁
         return true;
     }
     
     // time 是超时释放时间（单位：毫秒）
     // time = time - (System.currentTimeMillis() - current)
     time -= System.currentTimeMillis() - current; // 刷新超时释放时间
     if (time <= 0) { // 超时释放时间到了
         acquireFailed(waitTime, unit, threadId);
         return false; // 获取锁失败
     }
     
     // 获取 当前时间毫秒值
     current = System.currentTimeMillis();
     // 等待释放锁的通知（订阅释放锁的信号）
     RFuture subscribeFuture = subscribe(threadId);

 	// 等到超时释放时间结束还没有收到释放锁的通知的话, 返回 false
     // 获取锁失败 
     if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
         if (!subscribeFuture.cancel(false)) { // 取消订阅
             subscribeFuture.onComplete((res, e) -> {
                 if (e == null) {
                     unsubscribe(subscribeFuture, threadId);
                 }
             });
         }

     
         acquireFailed(waitTime, unit, threadId);
         return false;
     }
     
     // 接收到释放锁的信号
     try {
     	// 判断释放到超时时间
         time -= System.currentTimeMillis() - current;
         if (time <= 0) {
             acquireFailed(waitTime, unit, threadId);
             return false; // 获取锁失败
         }
     
         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) {
                 subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
             } else {
                 subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
             }

             time -= System.currentTimeMillis() - currentTime;
             if (time <= 0) {
                 acquireFailed(waitTime, unit, threadId);
                 return false;
             }
         }
     } finally {
         unsubscribe(subscribeFuture, threadId);
     }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
 }

  private RFuture tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
      // 判断超时释放时间是否是 -1
      if (leaseTime != -1) {
          return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
      }

      // 代码来到这里表示超时释放时间是 -1
      // tryLockInnerAsync 异步的方法
      RFuture ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                                  commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                                  TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
      ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
          if (e != null) {
              return;
          }

          // lock acquired
          if (ttlRemaining) {
              scheduleExpirationRenewal(threadId);
          }
      });
      return ttlRemainingFuture;
  }

tryLockInnerAsync()、tryAcquire()： ① 尝试获取锁。获取锁成功，返回 nil；获取锁失败，返回超时是否时间（lessTime）

② 该方法是异步的

① 获取锁失败后，不会立刻重新尝试获取锁

② 会等待释放锁的通知

五、Redisson 超时释放（锁的 ttl）

好😵懵逼啊，随便看看得了

📝 可重入：利用hash结构记录线程id和重入次数

📝 可重试：利用信号量和 PubSub 功能实现等待、唤醒，获取锁失败的重试机制

📝 超时续约：利用 watchDog，每隔一段时间（releaseTime / 3），重置超时时间

六、主从一致（连锁 MultiLock）

✏️主节点支持写请求

✏️从节点支持读请求

✏️主节点写入的数据要同步到从节点

✏️假如向主节点写入了数据，主节点还没有来得及向从节点同步数据自己就宕机了，此时 Redis 的哨兵机制会从剩下的从节点中选一个从节点作为新的主节点

✏️ 该新的主节点是没有上个主节点中的 lock = thread1 数据的

✏️ 锁失效，可能存在线程安全问题

✏️其中有任何一个节点宕机都无法拿到锁

七、锁总结

🎄不可重入 Redis 分布式锁：

原理： 利用 setnx 的互斥性；利用 ex 避免死锁；释放锁时判断线程标

缺陷： 不可重入、无法重试、锁超时失效

🎄可重入的 Redis 分布式锁：

原理： 利用 Hash 结构，记录线程标识和重入次数；利用 watchDog 延续锁过期时间；利用信号量控制锁重试等待

缺陷： redis宕机引起锁失效问题

🎄 Redisson 的 multiLock：

原理： 多个独立的 Redis 节点，必须在所有节点都获取重入锁，才算获取锁成功

缺陷： 运维成本高、实现复杂