Redisson

by emanjusaka from https://www.emanjusaka.top/2024/03/redisson-distributed-lock 彼岸花开可奈何本文欢送分享与聚合，全文转载请留下原文地址。实现分布式锁通常有三种形式：数据库、Redis 和 Zookeeper。咱们比拟罕用的是通过 Redis 和 Zookeeper 实现分布式锁。Redisson 框架中封装了通过 Redis 实现的分布式锁，上面咱们剖析一下它的具体实现。 关键点原子性 要么都胜利，要么都失败 过期工夫 如果锁还没来得及开释就遇到了服务宕机，就会呈现死锁的问题。给 Redis 的 key 设置过期工夫，即便服务宕机了超过设置的过期工夫锁会主动进行开释。 锁续期 因为给锁设置了过期工夫而咱们的业务逻辑具体要执行多长时间可能是变动和不确定的，如果设定了一个固定的过期工夫，可能会导致业务逻辑还没有执行完，锁被开释了的问题。锁续期能保障锁是在业务逻辑执行完才被开释。 正确开释锁 保障开释本人持有的锁，不能呈现 A 开释了 B 持有锁的状况。 Redis 实现分布式锁的几种部署形式单机 在这种部署形式中，Redis 的所有实例都部署在同一台服务器上。这种部署形式简单易行，但存在单点故障的危险。如果 Redis 实例宕机，则所有分布式锁都将生效。 哨兵 在这种部署形式中，Redis 的多个实例被配置为哨兵。哨兵负责监控 Redis 实例的状态，并在主实例宕机时主动选举一个新的主实例。这种部署形式能够提供更高的可用性和容错性。 集群 在这种部署形式中，Redis 的多个实例被配置为一个集群。集群中的每个实例都是平等的，并且能够解决读写操作。这种部署形式能够提供最高的可用性和容错性。 红锁 搞几个独立的 Master，比方 5 个，而后挨个加锁，只有超过一半以上（这里是 5/2+1=3 个）就代表加锁胜利，而后开释锁的时候也逐台开释。 应用形式引入依赖 <!-- pom.xml文件--><dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter</artifactId> <version>3.17.7</version></dependency>版本依赖： redisson-spring-data module nameSpring Boot versionredisson-spring-data-161.3.yredisson-spring-data-171.4.yredisson-spring-data-181.5.yredisson-spring-data-2x2.x.yredisson-spring-data-3x3.x.yyml配置 spring: redis: redisson: config: singleServerConfig: address: redis://127.0.0.1:6379 database: 0 password: null timeout: 3000间接注入应用 ...

一、前言分布式锁在理论工作中的利用还是比拟多的，其实现形式也有很多种，常见的有基于数据库锁、基于zookeeper、基于redis的，明天咱们来讲下基于redis实现的分布式锁。 redisson是一个redis客户端框架，提供了分布式锁的性能个性，这里咱们通过解析redisson的源码来剖析它是如何基于redis来实现分布式锁的？ 二、源码解析2.1 样例代码上面是一个分布式锁的简略样例代码 // 初始化配置，创立Redisson客户端Config config = new Config();config.setCodec(new JsonJacksonCodec()) .useSingleServer() .setAddress("redis://192.168.10.131:6379");RedissonClient client = Redisson.create(config);// 获取分布式锁RLock lock = client.getLock("myLock");lock.lock();System.out.println(Thread.currentThread().getId() + ": 获取到分布式锁");try { Thread.sleep(60 * 1000);} catch (Exception e) { e.printStackTrace();} finally { // 解锁 lock.unlock();}下面的样例代码比较简单，通过redisson客户端获取一个分布式锁，该分布式锁的key为myLock，睡眠60秒之后开释锁。这里比拟重要的是lock()办法，该办法是获取锁的具体步骤，所以接下来具体解析一下该办法 2.2 整体流程获取锁的流程图如下 具体流程为： 第一次尝试获取锁，如果获取到锁，间接返回。如果未获取到锁，返回锁的残余过期工夫ttl当未获取到锁时，订阅频道redisson_lock__channel:{myLock}（订阅该频道的作用是，当该分布式锁被其余拥有者所开释时，会往该订阅频道发送一个解锁音讯UNLOCK_MESSAGE，这时以后期待该分布式锁的线程会中断期待，并再次尝试获取锁）开启死循环，尝试获取锁，如果未获取到锁，拿到锁的残余过期工夫，并期待该锁的残余过期工夫（两头过程中如果订阅频道有解锁音讯UNLOCK_MESSAGE，会提前中断期待，持续循环），直到获取锁，退出循环获取到锁之后，勾销订阅频道源码如下 private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException { long threadId = Thread.currentThread().getId(); // 1、第一次尝试获取锁，ttl为null，示意获取到锁，间接return Long ttl = tryAcquire(-1, leaseTime, unit, threadId); if (ttl == null) { return; } // 2、订阅频道redisson_lock__channel:{myLock} CompletableFuture<RedissonLockEntry> future = subscribe(threadId); pubSub.timeout(future); RedissonLockEntry entry; if (interruptibly) { entry = commandExecutor.getInterrupted(future); } else { entry = commandExecutor.get(future); } try { // 3、开启循环 while (true) { // 再次尝试获取锁，ttl为null，示意获取到锁，退出循环 ttl = tryAcquire(-1, leaseTime, unit, threadId); if (ttl == null) { break; } // 如果ttl大于等于0 if (ttl >= 0) { try { // 期待ttl工夫 或者 接管到解锁音讯 entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { ... } } else { // 如果ttl小于0，阐明该锁未设置过期工夫，期待接管解锁音讯 if (interruptibly) { entry.getLatch().acquire(); } else { entry.getLatch().acquireUninterruptibly(); } } } } finally { // 退出订阅频道redisson_lock__channel:{myLock} unsubscribe(entry, threadId); } }2.3 锁的获取那么如何示意以后线程获取到锁？ ...