当前位置： 首页 > news >正文

Redission分布式锁详解

news 2024/10/22 22:02:47

前言


​ 在分布式系统中，当不同进程或线程一起访问共享资源时，会造成资源争抢，如果不加以控制的话，就会引发程序错乱。而分布式锁它采用了一种互斥机制来防止线程或进程间相互干扰，从而保证了数据的一致性。

常见的分布式锁实现方案

  • 基于 Redis 实现分布式锁
  • 基于 Zookeeper 实现分布式锁

介绍


​ **Redission是一个基于Redis实现的Java分布式对象存储和缓存框架。它提供了丰富的分布式数据结构和服务。**例如：分布式锁、分布式队列、分布式Rate Limiter等。

分布式锁的特点

  • 互斥性是分布式锁的重要特点，在任意时刻，只有一个线程能够持有锁
  • **锁的超时时间。**一个线程在持锁期间挂掉了而没主动释放锁，此时通过超时时间来保证该线程在超时后可以释放锁，这样其他线程才可以继续获取锁；
  • 加锁和解锁必须是由同一个线程来设置
  • Redis 是缓存型数据库，拥有很高的性能，因此加锁和释放锁开销较小，并且能够很轻易地实现分布式锁。

特性


高性能
​ Redission是基于Redis的，因此它继承了Redis的高性能和低延迟的特性。同时，它采用了Netty的NIO框架，能够并发地处理大量的请求，使得应用程序的响应速度得到了极大的提升。

易用性
​ Redission提供了丰富的API和方法，同时还提供了文档和示例，让开发者易于上手和使用。此外，它支持自动配置和灵活的配置方式，使得开发者可以根据自己的需求进行配置和调整。

可扩展性
​ Redission的分布式架构使得它支持水平扩展，可以将数据和请求分散到更多的节点上进行处理。这也使得它具备了更好的容错能力和可靠性。

常用场景


缓存
​ Redission支持不同的数据存储类型，例如：String、List、Set、Map、BloomFilter、HyperLogLog等，它可以将这些数据存储在Redis中，以实现数据缓存的功能，从而提高应用程序的性能和响应速度。

分布式锁
​ Redission提供了可重入锁、公平锁等常用的分布式锁，还支持异步执行、锁的自动续期、锁的等待等特性。

分布式队列
​ Redission提供了分布式队列的实现，我们可以在不同的节点之间快速、可靠地实现任务的传递和处理。

Redis分布式锁命令
常用命令

​ Redis 分布式锁常用命令如下所示：

  • SETNX key val：仅当key不存在时，设置一个 key 为 value 的字符串，返回1；若 key 存在，设置失败，返回 0；
  • Expire key timeout：为 key 设置一个超时时间，以 second 秒为单位，超过这个时间锁会自动释放，避免死锁；
  • DEL key：删除 key。

上述 SETNX 命令相当于抢占锁操作，EXPIRE 是为避免出现意外用来设置锁的过期时间，也就是说到了指定的过期时间，该客户端必须让出锁，让其他客户端去持有。

​ 但还有一种情况，如果在 SETNX 和 EXPIRE 之间服务器进程突然挂掉，也就是还未设置过期时间，这样就会导致 EXPIRE 执行不了，因此还是会造成“死锁”的问题。为了避免这个问题，Redis 作者在 2.6.12 版本后，对 SET 命令参数做了扩展，使它可以同时执行 SETNX 和 EXPIRE 命令，从而解决了死锁的问题。

直接使用 SET 命令实现，语法格式如下：

SET key value [expiration EX seconds|PX milliseconds] [NX|XX]  

  • EX second：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
  • PX millisecond：设置键的过期时间为毫秒。SET key value PX millisecond 效果等同于 PSETEX key millisecondvalue 。
  • NX：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
  • XX：只在键已经存在时，才对键进行设置操作。

 

原理


使用
使用redisson实现分布式锁的操作步骤，三部曲

  • 第一步： 获取锁 RLock redissonLock = redisson.getLock(lockKey);
  • 第二步： 加锁，实现锁续命功能 redissonLock.lock();
  • 第三步： 释放锁 redissonLock.unlock();


Redis 实现分布式锁主要步骤

  1. 指定一个 key 作为锁标记，存入 Redis 中，指定一个 唯一的用户标识 作为 value。
  2. 当 key 不存在时才能设置值，确保同一时间只有一个客户端进程获得锁，满足 互斥性 特性。
  3. 设置一个过期时间，防止因系统异常导致没能删除这个 key，满足 防死锁 特性。
  4. 当处理完业务之后需要清除这个 key 来释放锁，清除 key 时需要校验 value 值，需要满足 只有加锁的人才能释放锁 。

注意：

​ 以上实现步骤考虑到了使用分布式锁需要考虑的互斥性、防死锁、加锁和解锁必须为同一个进程等问题，但是锁的续期无法实现。所以通常情况都是采用 Redisson 实现 Redis 的分布式锁，借助 Redisson 的 WatchDog 机制 能够很好的解决锁续期的问题。

​ Watch Dog 机制其实就是一个后台定时任务线程，获取锁成功之后，会将持有锁的线程放入到一个 RedissonLock.EXPIRATION_RENEWAL_MAP里面，然后每隔 10 秒 （internalLockLeaseTime / 3） 检查一下，如果客户端 1 还持有锁 key（判断客户端是否还持有 key，其实就是遍历 EXPIRATION_RENEWAL_MAP 里面线程 id 然后根据线程 id 去 Redis 中查，如果存在就会延长 key 的时间），那么就会不断的延长锁 key 的生存时间

Redisson的锁机制
1）加锁机制

​ 加锁其实是通过一段 lua 脚本实现的。这里 KEYS[1] 代表的是你加锁的 key，比如你自己设置了加锁的那个锁 key 就是 “myLock”。

if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hincrby', 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]);"

 

ARGV[1] 代表的是锁 key 的默认生存时间，默认 30 秒。ARGV[2] 代表的是加锁的客户端的 ID，类似于下面这样：285475da-9152-4c83-822a-67ee2f116a79:52。至于最后面的一个 1 是为了后面可重入做的计数统计.

​ 第一段 if 判断语句，就是用 exists myLock 命令判断一下，如果你要加锁的那个锁 key 不存在的话，你就进行加锁。如何加锁呢？使用 hincrby 命令设置一个 hash 结构。接着会执行 pexpire myLock 30000 命令，设置 myLock 这个锁 key 的生存时间是 30 秒。到此为止，加锁完成。

2）锁互斥机制


​ 第二个 if 判断，判断一下，myLock 锁 key 的 hash 数据结构中，是否包含客户端 2 的 ID，这里明显不是，因为那里包含的是客户端 1 的 ID。所以，客户端 2 会执行：

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

返回的一个数字，这个数字代表了 myLock 这个锁 key 的剩余生存时间。

3）Watch dog 机制


​ 主要用于锁续费服务。只要客户端 1 一旦加锁成功，就会启动一个 Watch Dog。也就是说leaseTime 必须是 -1 才会开启 Watch Dog 机制，也就是如果你想开启 Watch Dog 机制必须使用默认的加锁时间为 30s。

4）可重入加锁机制


​ 当锁key已经存在，第二个 if 判断会成立，因为 myLock 的 hash 数据结构中包含的那个 ID 即客户端 1 的 ID，此时就会执行可重入加锁的逻辑，

5）锁释放机制

  1. 删除锁。
  2. 广播释放锁的消息，通知阻塞等待的进程（向通道名为 redisson_lock__channel publish 一条 UNLOCK_MESSAGE 信息）。
  3. 取消 Watch Dog 机制，即将 RedissonLock.EXPIRATION_RENEWAL_MAP 里面的线程 id 删除，并且 cancel 掉 Netty 的那个定时任务线程。


源码分析


​ 重点主要是依赖lua脚本的原子性，实现加锁和释放锁的功能。

实例化RedissonLock

  • super(commandExecutor, name); 父类name赋值，后续通过getName()获取
  • commandExecutor： 执行lua脚本的executor
  • id 是个UUID， 后面被用来当做和threadId组成 value值**，用作判断加锁和释放锁是否是同一个线程的校验。**
  • internalLockLeaseTime : 取自 Config#lockWatchdogTimeout，默认30秒，这个参数还有另外一个作用，锁续命的执行周期 internalLockLeaseTime/3 = 10秒


加锁和锁的续命redissonLock.lock()

  • **Thread.currentThread().interrupt()。**当发生异常，则通知线程进行中断并释放锁。
  • lockInterruptibly(-1, null); 此环节会先获取当前线程的线程ID，之后会尝试获取锁的剩余时间。
  • ​ 如果当前锁的剩余时间为null,说明没有线程持有该锁，直接返回让当前线程加锁成功。如果当前线程的剩余时间不为null,就会一直尝试获取锁。
  • ​ Redisson 提供了一个续期机制， 只要客户端 1 一旦加锁成功，就会启动一个 Watch Dog。**也就是说leaseTime 必须是 -1 才会开启 Watch Dog 机制，也就是如果你想开启 Watch Dog 机制必须使用默认的加锁时间为 30s。**如果你自己自定义时间，超过这个时间，锁就会自定释放，并不会延长。
  • **tryAcquireAsync(leaseTime, unit, threadId)（自旋获取锁的方法）。**其内部专门创建异步任务用于尝试获取锁。其任务内部会注册监听事件，当剩余时间为null，就会再次去获取锁，并给锁延长过期时间。
  • tryLockInnerAsync 其内部实现是lua脚本。
  • scheduleExpirationRenewal(final long threadId)。又是lua脚本 判断是否存在，存在就调用pexpire

释放锁redissonLock.unlock()

  • unlock()。先获取锁状态，如果锁状态为null，则抛出异常。否则就释放锁，并删除key。
  • unlockInnerAsync(long threadId)。unlock的内部实现，也是lua脚本，用于获取当前线程的锁状态。

Redissson tryLock源码

@Overridepublic boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {long time = unit.toMillis(waitTime);long current = System.currentTimeMillis();long threadId = Thread.currentThread().getId();// 1.尝试获取锁Long ttl = tryAcquire(leaseTime, unit, threadId);// lock acquiredif (ttl == null) {return true;}// 申请锁的耗时如果大于等于最大等待时间，则申请锁失败.time -= System.currentTimeMillis() - current;if (time <= 0) {acquireFailed(threadId);return false;}current = System.currentTimeMillis();/*** 2.订阅锁释放事件，并通过 await 方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题：* 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，一旦锁释放会发消息通知待等待的线程进行竞争.** 当 this.await 返回 false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败.* 当 this.await 返回 true，进入循环尝试获取锁.*/RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);// await 方法内部是用 CountDownLatch 来实现阻塞，获取 subscribe 异步执行的结果（应用了 Netty 的 Future）if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {if (!subscribeFuture.cancel(false)) {subscribeFuture.onComplete((res, e) -> {if (e == null) {unsubscribe(subscribeFuture, threadId);}});}acquireFailed(threadId);return false;}try {// 计算获取锁的总耗时，如果大于等于最大等待时间，则获取锁失败.time -= System.currentTimeMillis() - current;if (time <= 0) {acquireFailed(threadId);return false;}/*** 3.收到锁释放的信号后，在最大等待时间之内，循环一次接着一次的尝试获取锁* 获取锁成功，则立马返回 true，* 若在最大等待时间之内还没获取到锁，则认为获取锁失败，返回 false 结束循环*/while (true) {long currentTime = System.currentTimeMillis();// 再次尝试获取锁ttl = tryAcquire(leaseTime, unit, threadId);// lock acquiredif (ttl == null) {return true;}// 超过最大等待时间则返回 false 结束循环，获取锁失败time -= System.currentTimeMillis() - currentTime;if (time <= 0) {acquireFailed(threadId);return false;}/*** 6.阻塞等待锁（通过信号量(共享锁)阻塞,等待解锁消息）：*/currentTime = System.currentTimeMillis();if (ttl >= 0 && ttl < time) {//如果剩余时间(ttl)小于wait time ,就在 ttl 时间内，从Entry的信号量获取一个许可(除非被中断或者一直没有可用的许可)。getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {//则就在wait time 时间范围内等待可以通过信号量getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);}// 更新剩余的等待时间(最大等待时间-已经消耗的阻塞时间)time -= System.currentTimeMillis() - currentTime;if (time <= 0) {acquireFailed(threadId);return false;}}} finally {// 7.无论是否获得锁,都要取消订阅解锁消息unsubscribe(subscribeFuture, threadId);}
//        return get(tryLockAsync(waitTime, leaseTime, unit));}

流程分析：

  1. 尝试获取锁，返回 null 则说明加锁成功，返回一个数值，则说明已经存在该锁，ttl 为锁的剩余存活时间。
  2. 如果此时客户端 2 进程获取锁失败，那么使用客户端 2 的线程 id（其实本质上就是进程 id）通过 Redis 的 channel 订阅锁释放的事件，。如果等待的过程中一直未等到锁的释放事件通知，当超过最大等待时间则获取锁失败，返回 false，也就是第 39 行代码。如果等到了锁的释放事件的通知，则开始进入一个不断重试获取锁的循环。
  3. 循环中每次都先试着获取锁，并得到已存在的锁的剩余存活时间。如果在重试中拿到了锁，则直接返回。如果锁当前还是被占用的，那么等待释放锁的消息，具体实现使用了 JDK 的信号量 Semaphore 来阻塞线程，当锁释放并发布释放锁的消息后，信号量的 release() 方法会被调用，此时被信号量阻塞的等待队列中的一个线程就可以继续尝试获取锁了。

特别注意：

​ 以上过程存在一个细节，这里有必要说明一下，也是分布式锁的一个关键点：当锁正在被占用时，等待获取锁的进程并不是通过一个 while(true) 死循环去获取锁，而是利用了 Redis 的发布订阅机制,通过 await 方法阻塞等待锁的进程，有效的解决了无效的锁申请浪费资源的问题。

分布式锁失效问题
1、事务相关
问题描述：
​ 加锁、解锁环节放在事务注解修饰的方法内部

数据执行流程：
​ lock上锁-》开启事务-》查询释放满足条件（业务方面）-》执行业务处理-》unlock解锁-》提交事务

解决方案一（多事务）：
​ 事务内含多层事务，加锁、释放锁放事务外层， 若是释放锁失败的话，则进行数据回滚。

解决方案二（单事务）：
单事务内可以加锁 释放锁，释放锁采用以下方法。

 TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {

    @Override
    public void afterCompletion(int status) {
        copyLock.unlock();
        log.info("事务结束自动释放分布式锁成功key:{}", copyKey);
    }
});

2、高可用解决方案
问题描述：


​ 主从复制架构，在主节点异步同步从节点过程中，主节点宕机，那么分布式锁的数据就会丢失，即使后续从节点顶上，也是未加锁状态，也是可以被其他线程抢到的。

解决方案1：
​ 1、在并发场景不严重的情况下，可以选择Zookeeper方案

​ 2、在释放锁环节，抛出异常那里，添加数据回溯处理。若是释放锁失败，说明当前锁处于无锁状态或者被别的线程拿到锁（主从同步宕机），那么当前数据的操作就处于无效状态，则需要进行回撤处理。

解决方案2：
​ 采用红锁来解决。比如1主4从

​ 即在代码中不依赖于主从，将这5台机器视为平等的，在代码中依次对这5台机器去加锁，只有成功的机器数大于一半就算加锁成功，其他机器也就没必要再去操作了，相反，如果大于一半的机器失败了，就算失败，其他机器也就没必要再去操作了。

优缺点
优点
Redisson 通过 Watch Dog 机制很好的解决了锁的续期问题。
和 Zookeeper 相比较，Redisson 基于 Redis 性能更高，适合对性能要求高的场景。
通过 Redisson 实现分布式可重入锁，比原生的 SET mylock userId NX PX milliseconds + lua 实现的效果更好些，虽然基本原理都一样，但是它帮我们屏蔽了内部的执行细节。
在等待申请锁资源的进程等待申请锁的实现上也做了一些优化，减少了无效的锁申请，提升了资源的利用率。
缺点
1、不能使用自定义过期时间，不然会影响Redisson无法正常续费问题。

总结
​ 整体来说，使用Redisson可以很简单的集成到项目中，并且使用方便，性能非常好，可以有效解决争取共享资源有序问题。

 


http://www.mrgr.cn/news/56267.html

相关文章：

  • Android 从0搭建初始化MVVM项目框架（二）：添加版本依赖管理、分包分模块、组件化Aroute
  • 数字化变革：不在技术，而在组织
  • 【力扣刷题实战】有效的括号
  • 【网络原理】HTTP协议
  • C语言 | Leetcode C语言题解之第493题翻转对
  • 【FAQ】HarmonyOS SDK 闭源开放能力 —Map Kit（3）
  • YOLO目标检测理论详解，YOLOv1理论知识讲解，超w字精读（学习YOLO框架必备），全网最详细教程
  • 嵌入式 MCU 编程提速秘籍 —— 让你的程序飞起来！
  • 书生实战营第四期-第一关Linux+InternStudio
  • 肉桂酰辅酶A还原酶Cinnamoyl-CoA Reductases（CCR）表征及晶体-文献精读70
  • 0基础学java之Day12
  • 二分查找_在排序数组中查找元素的第一个和最后一个位置
  • 超详细JDK安装+环境配置教程
  • vnc+wsl2试用
  • 深入浅出剖析重量级文生图模型Flux.1
  • 数据结构图的应用最小生成树-普里姆算法(C语言代码+无向网+有向网+邻接矩阵存储结构)-最低附带图片+终端输入内容方便理解
  • 【Python爬虫系列】_031.Scrapy_模拟登陆中间件
  • 让你的 IDEA 使用更流畅 | IDEA内存修改
  • 常见的加密算法的分类及其原理
  • 利用自定义 ref 实现函数防抖
  • 批量合并同名Labelme标注文件内容
  • freeRTOS中互斥锁与信号量使用？
  • vue3学习记录-v-model
  • Numpy基础02
  • 浏览器控制的无线开关
  • 【03】RabbitMQ核心功能扩展