Redis分布式可重入锁实现方案

前言

在单进程环境下，要保证一个代码块的同步执行，直接用synchronized 关键字或ReetrantLock 即可。在分布式环境下，要保证多个节点的线程对代码块的同步访问，就必须要用到分布式锁方案。
分布式锁实现方案有很多，有基于关系型数据库行锁实现的；有基于ZooKeeper临时顺序节点实现的；还有基于 Redis setnx 命令实现的。本文介绍一下基于 Redis 实现的分布式锁方案。

理解分布式锁

实现分布式锁有几个要求

  • 互斥性：任意时刻，最多只会有一个客户端线程可以获得锁
  • 可重入：同一客户端的同一线程，获得锁后能够再次获得锁
  • 避免死锁：客户端获得锁后即使宕机，后续客户端也可以获得锁
  • 避免误解锁：客户端A加的锁只能由A自己释放
  • 释放锁通知：持有锁的客户端释放锁后，最好可以通知其它客户端继续抢锁
  • 高性能和高可用

Redis 服务端命令是单线程串行执行的，天生就是原子的，并且支持执行自定义的 lua 脚本，功能上更加强大。
关于互斥性，我们可以用 setnx 命令实现，Redis 可以保证只会有一个客户端 set 成功。但是由于我们要实现的是一个分布式的可重入锁，数据结构得用 hash，用客户端ID+线程ID作为 field，value 记作锁的重入次数即可。
关于死锁，代码里建议把锁的释放写在 finally 里面确保一定执行，针对客户端抢到锁后宕机的场景，可以给 redis key 设置一个超时时间来解决。
关于误解锁，客户端在释放锁时，必须判断 field 是否当前客户端ID以及线程ID一致，不一致就不执行删除，这里需要用到 lua 脚本判断。
关于释放锁通知，可以利用 Redis 发布订阅模式，给每个锁创建一个频道，释放锁的客户端负责往频道里发送消息通知等待抢锁的客户端。
最后关于高性能和高可用，因为 Redis 是基于内存的，天生就是高性能的。但是 Redis 服务本身一旦出现问题，分布式锁也就不可用了，此时可以多部署几台独立的示例，使用 RedLock 算法来解决高可用的问题。

设计实现

首先我们定义一个 RedisLock 锁对象的抽象接口，只有尝试加锁和释放锁方法

public interface RedisLock {boolean tryLock();boolean tryLock(long waitTime, long leaseTime, TimeUnit unit);void unlock();
}

然后提供一个默认实现 DefaultRedisLock

public class DefaultRedisLock implements RedisLock {// 客户端ID UUIDprivate final String clientId;private final StringRedisTemplate redisTemplate;// 锁频道订阅器 接收释放锁通知private final LockSubscriber lockSubscriber;// 加锁的keyprivate final String lockKey;
}

关于tryLock() ，首先执行lua脚本尝试获取锁，如果加锁失败则返回其它客户端持有锁的过期时间，客户端订阅锁对应的频道，然后sleep，直到收到锁释放的通知再继续抢锁。最终不管有没有抢到锁，都会在 finally 取消频道订阅。

@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) {final long timeout = System.currentTimeMillis() + unit.toMillis(waitTime);final long threadId = Thread.currentThread().getId();Long ttl = tryAcquire(leaseTime, unit, threadId);if (ttl == null) {return true;}if (System.currentTimeMillis() >= timeout) {return false;}final Semaphore semaphore = lockSubscriber.subscribe(getChannel(lockKey), threadId);try {while (true) {if (System.currentTimeMillis() >= timeout) {return false;}ttl = tryAcquire(leaseTime, unit, threadId);if (ttl == null) {return true;}if (System.currentTimeMillis() >= timeout) {return false;}semaphore.tryAcquire(timeout - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}} catch (Exception e) {e.printStackTrace();} finally {lockSubscriber.unsubscribe(getChannel(lockKey), threadId);}return false;
}

tryAcquire() 就是执行lua脚本来加锁，解释一下这段脚本的逻辑：首先判断 lockKey 是否存在，不存在则直接设置 lockKey并且设置过期时间，返回空，表示加锁成功。存在则判断 field 是否和当前客户端ID+线程ID一致，一致则代表锁重入，递增一下value即可，不一致代表加锁失败，返回锁的过期时间

private Long tryAcquire(long leaseTime, TimeUnit timeUnit, long threadId) {return redisTemplate.execute(RedisScript.of("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]);", Long.class), Collections.singletonList(lockKey),String.valueOf(timeUnit.toMillis(leaseTime)), getLockName(threadId));
}

lockName是由客户端ID和线程ID组成的：

private String getLockName(long threadId) {return clientId + ":" + threadId;
}

如果加锁失败，客户端会尝试订阅对应的频道，名称规则是：

private String getChannel(String lockKey) {return "__lock_channel__:" + lockKey;
}

订阅方法是LockSubscriber#subscribe ，同一个频道无需订阅多个监听器，所以用一个 Map 记录。订阅成功以后，会返回当前线程对应的一个 Semaphore 对象，默认许可数是0，当前线程会调用Semaphore#tryAcquire 等待许可数，监听器在收到锁释放消息后会给 Semaphore 对象增加许可数，唤醒线程继续抢锁。

@Component
public class LockSubscriber {@Autowiredprivate RedisMessageListenerContainer messageListenerContainer;private final Map<String, Map<Long, Semaphore>> channelSemaphores = new HashMap<>();private final Map<String, MessageListener> listeners = new HashMap<>();private final StringRedisSerializer serializer = new StringRedisSerializer();public synchronized Semaphore subscribe(String channelName, long threadId) {MessageListener old = listeners.put(channelName, new MessageListener() {@Overridepublic void onMessage(Message message, byte[] pattern) {String channel = serializer.deserialize(message.getChannel());String ignore = serializer.deserialize(message.getBody());Map<Long, Semaphore> semaphoreMap = channelSemaphores.get(channel);if (semaphoreMap != null && !semaphoreMap.isEmpty()) {semaphoreMap.values().stream().findFirst().ifPresent(Semaphore::release);}}});if (old == null) {messageListenerContainer.addMessageListener(listeners.get(channelName), new ChannelTopic(channelName));}Semaphore semaphore = new Semaphore(0);Map<Long, Semaphore> semaphoreMap = channelSemaphores.getOrDefault(channelName, new HashMap<>());semaphoreMap.put(threadId, semaphore);channelSemaphores.put(channelName, semaphoreMap);return semaphore;}public synchronized void unsubscribe(String channelName, long threadId) {Map<Long, Semaphore> semaphoreMap = channelSemaphores.get(channelName);if (semaphoreMap != null) {semaphoreMap.remove(threadId);if (semaphoreMap.isEmpty()) {MessageListener listener = listeners.remove(channelName);if (listener != null) {messageListenerContainer.removeMessageListener(listener);}}}}
}

对于 unlock，就只是一段 lua 脚本，这里解释一下：判断当前客户端ID+线程ID 这个 field 是否存在，存在说明是自己加的锁，可以释放。不存在说明不是自己加的锁，无需做任何处理。因为是可重入锁，每次 unlock 都只是递减一下 value，只有当 value 等于0时才是真正的释放锁。释放锁的时候会 del lockKey，再 publish 发送锁释放通知，让其他客户端可以继续抢锁。

@Override
public void unlock() {long threadId = Thread.currentThread().getId();redisTemplate.execute(RedisScript.of("if (redis.call('hexists', KEYS[1], ARGV[1]) == 0) then " +"return nil;end;" +"local counter = redis.call('hincrby', KEYS[1], ARGV[1], -1); " +"if (counter > 0) then " +"return 0; " +"else " +"redis.call('del', KEYS[1]); " +"redis.call('publish', KEYS[2], 1); " +"return 1; " +"end; " +"return nil;", Long.class), Arrays.asList(lockKey, getChannel(lockKey)),getLockName(threadId));
}

最后，我们需要一个 RedisLockFactory 来创建锁对象，它同时会生成客户端ID

@Component
public class RedisLockFactory {private static final String CLIENT_ID = UUID.randomUUID().toString();@Autowiredprivate StringRedisTemplate redisTemplate;@Autowiredprivate LockSubscriber lockSubscriber;public RedisLock getLock(String lockKey) {return new DefaultRedisLock(CLIENT_ID, redisTemplate, lockSubscriber, lockKey);}
}

至此，一个基于 Redis 实现的分布式可重入锁就完成了。

尾巴

目前这个版本的分布式锁，保证了互斥性、可重入、避免死锁和误解锁、实现了释放锁通知，但是并没有高可用的保证。如果 Redis 是单实例部署，就会存在单点问题，Redis 一旦故障，整个分布式锁将不可用。如果 Redis 是主从集群模式部署，虽然有主从自动切换，但是 Master 和 Slave 之间的数据同步是存在延迟的，分布式锁可能会出现问题。比如：客户端A加锁成功，lockKey 写入了 Master，此时 Master 宕机，其它 Slave 升级成了 Master，但是还没有同步到 lockKey，客户端B来加锁也会成功，这就没有保证互斥性。针对这个问题，可以参考 RedLock 算法，部署多个单独的 Redis 示例，只要一半以上的Redis节点加锁成功就算成功，来尽可能的保证服务高可用。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/688217.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

突破编程_C++_面试（变量与常量）

突破编程_C++_面试（变量与常量）

面试题 1 &#xff1a; C 中的变量存储类别有哪些&#xff0c;并简要描述它们的特点&#xff1f; 在C中&#xff0c;变量的存储类别决定了变量的生命周期和可见性。以下是C中的几种变量存储类别及其特点&#xff1a; 自动存储期 也称为局部存储类别。这类变量在函数或代码块…
阅读更多...
正大国际期货：日内交易

正大国际期货：日内交易

日内交易是一种交易模式&#xff0c;英文名字是daytrade,主要是指持仓时间短&#xff0c;不留过夜持仓的交易方式。日内交易捕捉入市后能够马上脱离入市成本的交易机会&#xff0c;入市之后如果不能马上获利&#xff0c;就准备迅速离场
阅读更多...
程序员可以做一辈子吗？大龄程序员出路在哪？

程序员可以做一辈子吗？大龄程序员出路在哪？

前言 随着2023年AI的出现&#xff0c;大家对待程序员工作有了一丝丝危机感&#xff0c;特别是今年整个IT行业进入了前所未有的寒冬期&#xff0c;让程序员不得不思考未来的职业发展。 甚至很多程序员一想到自己接近35岁&#xff0c;焦虑感油然而生&#xff0c;这也是大部分程…
阅读更多...
基于YOLOv7算法的高精度实时雾天车辆行人目标检测系统（PyTorch+Pyside6+YOLOv7）

基于YOLOv7算法的高精度实时雾天车辆行人目标检测系统（PyTorch+Pyside6+YOLOv7）

摘要&#xff1a;基于YOLOv7算法的高精度实时雾天车辆行人目标检测系统可用于日常生活中检测与定位bicycle、bus、car、motorbike和person&#xff0c;此系统可完成对输入图片、视频、文件夹以及摄像头方式的目标检测与识别&#xff0c;同时本系统还支持检测结果可视化与导出。…
阅读更多...
数据采集新纪元：Linux边缘计算技术在智慧工厂的应用解析

数据采集新纪元：Linux边缘计算技术在智慧工厂的应用解析

在当今全球智能制造的大潮下&#xff0c;Linux边缘计算网关正扮演着愈发重要的角色。它位于数据产生源头与云计算中心之间&#xff0c;为智慧工厂提供了关键的实时决策能力和高效的预测性维护解决方案。 以一家领先汽车零部件生产商为例&#xff0c;其高度自动化的生产线上的每…
阅读更多...
C/C++数据结构——剖析排序算法

C/C++数据结构——剖析排序算法

1. 排序的概念及其运用 1.1 排序的概念 https://en.wikipedia.org/wiki/Insertion_sorthttps://en.wikipedia.org/wiki/Insertion_sort 排序&#xff1a;所谓排序&#xff0c;就是使一串记录&#xff0c;按照其中的某个或某些关键字的大小&#xff0c;递增或递减的排列起来的…
阅读更多...
力扣hot2--哈希

力扣hot2--哈希

推荐博客&#xff1a; for(auto i : v)遍历容器元素_for auto 遍历-CSDN博客 字母异位词都有一个特点&#xff1a;也就是对这个词排序之后结果会相同。所以将排序之后的string作为key&#xff0c;将排序之后能变成key的单词组vector<string>作为value。 class Solution …
阅读更多...
探索未来科技前沿：深度学习的进展与应用

探索未来科技前沿：深度学习的进展与应用

深度学习的进展 摘要&#xff1a;深度学习作为人工智能领域的重要分支&#xff0c;近年来取得了巨大的进展&#xff0c;并在各个领域展现出惊人的应用潜力。本文将介绍深度学习的发展历程、技术原理以及在图像识别、自然语言处理等领域的应用&#xff0c;展望深度学习在未来的…
阅读更多...
SQL29 计算用户的平均次日留存率（lead函数的用法）

SQL29 计算用户的平均次日留存率（lead函数的用法）

代码 with t1 as(select distinct device_id,date --去重防止单日多次答题的情况from question_practice_detail ) select avg(if(datediff(date2,date1)1,1,0)) as avg_ret from (selectdistinct device_id,date as date1,lead(date) over(partition by device_id order by d…
阅读更多...
Effective Objective-C 学习（三）

Effective Objective-C 学习（三）

理解引用计数 Objective-C 使用引用计数来管理内存&#xff1a;每个对象都有个可以递增或递减的计数器。如果想使某个对象继续存活&#xff0c;那就递增其引用计数&#xff1a;用完了之后&#xff0c;就递减其计数。计数变为 0时&#xff0c;就可以把它销毁。 在ARC中&#xf…
阅读更多...
C++入门学习（二十八）跳转语句—continue语句

C++入门学习（二十八）跳转语句—continue语句

当在循环中遇到continue语句时&#xff0c;它会跳过当前迭代剩余的代码块&#xff0c;并立即开始下一次迭代。这意味着continue语句用于跳过循环中特定的执行步骤&#xff0c;而不是完全终止循环。 直接看一下下面的代码更清晰&#xff1a; 与上一节的break语句可以做一下对比…
阅读更多...
什么是软件测试？软件测试的目的与原则是什么？

什么是软件测试？软件测试的目的与原则是什么？

&#x1f525; 交流讨论&#xff1a;欢迎加入我们一起学习&#xff01; &#x1f525; 资源分享&#xff1a;耗时200小时精选的「软件测试」资料包 &#x1f525; 教程推荐&#xff1a;火遍全网的《软件测试》教程 &#x1f4e2;欢迎点赞 &#x1f44d; 收藏 ⭐留言 &#x1…
阅读更多...
原创详解OpenAI Sora是什么？技术先进在哪里？能够带来什么影响？附中英文技术文档

原创详解OpenAI Sora是什么？技术先进在哪里？能够带来什么影响？附中英文技术文档

一&#xff1a;Sora是什么 Sora是一个文本到视频的模型&#xff0c;由美国的人工智能研究机构OpenAI开发。Sora可以根据描述性的文本提示&#xff0c;生成高质量的视频&#xff0c;也可以根据已有的视频&#xff0c;向前或向后延伸&#xff0c;生成更长的视频。 Sora的主要功…
阅读更多...
UPC训练赛二十/20240217

UPC训练赛二十/20240217

A:无穷力量 题目描述 2022年重庆突发山火让世界看到了中国一个又一个的感人事迹&#xff1a;战士们第一时间奔赴火场&#xff0c;志愿者们自发组成团队&#xff0c;为救火提供一切的可能的服务&#xff0c;人们自发输送物资&#xff0c;有的志愿者甚至几天几夜没有睡觉。每个…
阅读更多...
反射的作用

反射的作用

获取一个类里面所有的信息&#xff0c;获取到了之后&#xff0c;再执行其他的业务逻辑结合配置文件&#xff0c;动态的创建对象并调用方法 练习1&#xff1a; public class MyTest {public static void main(String[] args) throws IllegalAccessException, IOException {Stude…
阅读更多...
Swift 5.9 新 @Observable 对象在 SwiftUI 使用中的陷阱与解决

Swift 5.9 新 @Observable 对象在 SwiftUI 使用中的陷阱与解决

概览 在 Swift 5.9 中&#xff0c;苹果为我们带来了全新的可观察框架 Observation&#xff0c;它是观察者开发模式在 Swift 中的一个全新实现。 除了自身本领过硬以外&#xff0c;Observation 框架和 SwiftUI 搭配起来也能相得益彰&#xff0c;事倍功半。不过 Observable 对象…
阅读更多...
C++学习Day06之继承方式

C++学习Day06之继承方式

目录 一、程序及输出1.1 公共继承1.1.1 父类中公共成员&#xff0c;子类可以正常访问1.1.2 父类中保护成员&#xff0c;子类类外不可以访问1.1.3 父类中私有成员&#xff0c;子类无法访问 1.2 保护继承1.2.1 父类中公共权限 子类中变为 保护权限1.2.2 父类中保护权限 子类中变为…
阅读更多...
自定义Linux登录自动提示语

自定义Linux登录自动提示语

设置提示语的方式 在Linux系统中&#xff0c;可以通过修改几个特定的文件来实现在用户登录时自动弹出提示语。以下是几个常用的方法&#xff1a; 1. 修改/etc/issue文件&#xff1a; 这个文件用于显示本地登录前的提示信息 sudo vi /etc/issue在项目合作的时候&#xff0c;…
阅读更多...
平衡二叉树（AVL树）

平衡二叉树（AVL树）

定义&#xff1a; 左右子树高度之差不超过1左右子树都是平衡二叉树 平衡二叉树的增删操作都离不开二叉树的调整 二叉树调整 LL型&#xff1a;右旋 LR型&#xff1a;左旋右旋 RR型&#xff1a;左旋 RL型&#xff1a;右旋左旋
阅读更多...
从小红书笔记详情API看电商如何提升品牌影响力

从小红书笔记详情API看电商如何提升品牌影响力

从小红书笔记详情API来看&#xff0c;电商如何提升品牌影响力是一个复杂但至关重要的过程。首先&#xff0c;理解小红书平台和其用户群体的特点是关键。小红书是一个以用户分享和消费体验为主的社交媒体平台&#xff0c;用户群体主要是年轻、有购买力的女性。因此&#xff0c;电…
阅读更多...
最新文章

Copyright @ 2022~2023