常见分布式锁4:zookeeper 瞬时znode节点 + watcher监听机制，ChatGPT回复的解决死锁的方案

news/2024/11/23 12:55:38/

原文地址在这里

临时节点具备数据自动删除的功能。当client与ZooKeeper连接和session断掉时，相应的临时节点就会被删除。zk有瞬时和持久节点，瞬时节点不可以有子节点。会话结束之后瞬时节点就会消失，基于zk的瞬时有序节点实现分布式锁：

多线程并发创建瞬时节点的时候，得到有序的序列，序号最小的线程可以获得锁；

其他的线程监听自己序号的前一个序号。前一个线程执行结束之后删除自己序号的节点；

下一个序号的线程得到通知，继续执行；

以此类推，创建节点的时候，就确认了线程执行的顺序。

<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.4.14</version><exclusions><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions>
</dependency>

zk 的观察器只可以监控一次，数据发生变化之后可以发送给客户端，之后需要再次设置监控。exists、create、getChildren三个方法都可以添加watcher ，也就是在调用方法的时候传递true就是添加监听。注意这里Lock 实现了Watcher和AutoCloseable：

当前线程创建的节点是第一个节点就获得锁，否则就监听自己的前一个节点的事件：

/*** 自己本身就是一个 watcher，可以得到通知* AutoCloseable 实现自动关闭，资源不使用的时候*/
@Slf4j
public class ZkLock implements AutoCloseable, Watcher {
​private ZooKeeper zooKeeper;
​/*** 记录当前锁的名字*/private String znode;
​public ZkLock() throws IOException {this.zooKeeper = new ZooKeeper("localhost:2181",10000,this);}
​public boolean getLock(String businessCode) {try {//创建业务 根节点Stat stat = zooKeeper.exists("/" + businessCode, false);if (stat==null){zooKeeper.create("/" + businessCode,businessCode.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}
​//创建瞬时有序节点  /order/order_00000001znode = zooKeeper.create("/" + businessCode + "/" + businessCode + "_", businessCode.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
​//获取业务节点下 所有的子节点List<String> childrenNodes = zooKeeper.getChildren("/" + businessCode, false);//获取序号最小的（第一个）子节点Collections.sort(childrenNodes);String firstNode = childrenNodes.get(0);//如果创建的节点是第一个子节点，则获得锁if (znode.endsWith(firstNode)){return true;}//如果不是第一个子节点，则监听前一个节点String lastNode = firstNode;for (String node:childrenNodes){if (znode.endsWith(node)){zooKeeper.exists("/"+businessCode+"/"+lastNode,true);break;}else {lastNode = node;}}synchronized (this){wait();}return true;} catch (Exception e) {e.printStackTrace();}return false;}
​@Overridepublic void close() throws Exception {zooKeeper.delete(znode,-1);zooKeeper.close();log.info("我已经释放了锁！");}
​@Overridepublic void process(WatchedEvent event) {if (event.getType() == Event.EventType.NodeDeleted){synchronized (this){notify();}}}
}

这段代码实现了一个基于 ZooKeeper 的分布式锁，以下是它的实现步骤：

  1. 首先创建 ZooKeeper 客户端，并实现 Watcher 接口，在自身上注册监听器。

  2. 在 ZooKeeper 上创建一个业务根节点，例如 /businessCode，表示该业务下所有的分布式锁。

  3. 使用 ZooKeeper 的临时有序节点创建子节点，例如 /businessCode/businessCode_00001，表示当前节点占用了分布式锁，并记录在 znode 中。

  4. 获取业务节点下所有的子节点，并按节点名称排序。如果当前节点的名称是所有子节点中最小的，则获取分布式锁。

  5. 如果当前节点不是所有子节点中最小的，则监听前一个子节点的删除事件，等待前一个子节点释放锁。

  6. 当前一个子节点被删除时，重新执行第四步，即再次检查当前节点是否为所有子节点中最小的。

  7. 如果当前节点成功获取到锁，则执行业务操作；否则等待锁释放，重新获取锁。

  8. 业务操作完成后，执行 close() 方法释放锁，删除当前节点。

总体来说，这段代码实现了一种基本的分布式锁机制，通过 ZooKeeper 的临时有序节点和 Watcher 监听机制来实现。需要注意的是，此实现方式还可能存在死锁问题，当持有锁的节点出现网络故障或宕机时，会导致整个分布式锁失效。因此，在实际应用中，还需要综合考虑各种异常情况，确保分布式锁的正确性和高可用性。

 针对死锁ChatGPT3.5的解决方案：

那段代码中存在潜在的死锁问题，即如果持有锁的节点出现异常，可能导致整个分布式锁失效。为了解决这个问题，可以考虑以下两种方案：

  1. 引入超时机制

在获取锁的过程中，可以引入超时机制，例如在等待前一个节点释放锁的过程中，设定一个超时时间，如果超过了这个时间仍未收到通知，则当前节点主动删除自己的临时节点，释放锁，并尝试重新获取锁。

  1. 心跳检测机制

对于已经获取到分布式锁的节点，可以在一定时间间隔内向 ZooKeeper 发送心跳消息，以检测自身是否还处于活跃状态。如果该节点长时间没有发送心跳消息或者 ZooKeeper 无法接收到该节点的心跳消息，就认为该节点出现了异常，需要主动删除自身的临时节点，释放锁，并尝试重新获取锁。

在前面提到的两种方案中，我选择了第一种方案，即引入超时机制来解决死锁问题。具体实现方式如下：

在获取锁的过程中，我们可以增加一个等待时间的参数 waitTime，表示等待前一个节点释放锁的最长时间。在设置监听器之后，使用 CountDownLatch 等待前一个节点的删除事件通知，当出现以下情况之一时，即可跳出等待循环：

  1. 在等待时间内成功获取到锁，即当前节点为业务节点下所有临时节点中的最小节点。

  2. 在等待时间内等待到超时时间，说明前一节点无法正常释放锁，因此当前节点放弃等待，主动删除自己的临时节点，重新尝试获取锁。

修改后的代码如下所示：

public class DistributedLock implements AutoCloseable {private static final String ROOT_NODE = "/businessCode"; // 业务根节点private final ZooKeeper zkClient; // ZooKeeper 客户端private final String lockName; // 分布式锁名称private String currentLockNodePath; // 当前节点的 znode 路径public DistributedLock(String zkServers, String lockName) throws IOException {this.zkClient = new ZooKeeper(zkServers, 3000, event -> {});this.lockName = lockName;ensureRootNodeExists();}// 确保业务根节点存在private void ensureRootNodeExists() {try {if (zkClient.exists(ROOT_NODE, false) == null) {zkClient.create(ROOT_NODE, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}} catch (Exception e) {throw new RuntimeException("Failed to ensure root node exists: " + ROOT_NODE, e);}}// 获取锁public void lock(long waitTime) {try {// 创建当前节点currentLockNodePath = zkClient.create(ROOT_NODE + "/" + lockName + "_",null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);// 等待前一个节点释放锁List<String> children = zkClient.getChildren(ROOT_NODE, false);String[] sortedNodes = children.stream().sorted().toArray(String[]::new);int currentNodeIndex = Arrays.binarySearch(sortedNodes, currentLockNodePath.substring(currentLockNodePath.lastIndexOf("/") + 1));if (currentNodeIndex == 0) {// 当前节点为业务节点下所有临时节点中的最小节点，获取锁return;}// 等待前一个节点的删除事件通知，如果超时仍未收到，则主动删除自己的临时节点CountDownLatch latch = new CountDownLatch(1);String previousNodePath = ROOT_NODE + "/" + sortedNodes[currentNodeIndex - 1];Stat stat = zkClient.exists(previousNodePath, event -> {if (event.getType() == Watcher.Event.EventType.NodeDeleted) {latch.countDown();}});if (stat != null) {if (latch.await(waitTime, TimeUnit.MILLISECONDS)) {// 等待成功，重新执行获取锁操作lock(waitTime);} else {// 超时未等到通知，主动删除自己的临时节点，放弃获取锁zkClient.delete(currentLockNodePath, -1);throw new RuntimeException("Failed to acquire lock within wait time: " + waitTime);}}} catch (Exception e) {throw new RuntimeException("Failed to acquire lock: " + lockName, e);}}// 释放锁@Overridepublic void close() {try {System.out.println("Releasing lock: " + lockName);zkClient.delete(currentLockNodePath, -1);zkClient.close();} catch (Exception e) {throw new RuntimeException("Failed to release lock: " + lockName, e);}}
}

在 lock() 方法中增加了一个参数 waitTime，表示等待前一个节点释放锁的最长时间。在方法最后，如果超时未收到通知，则主动删除当前节点，放弃获取锁，并抛出异常提示用户。

需要注意的是，在使用该类时，需要确保所有节点设置的 waitTime 值是一致的。另外，为了让线程能够跳出等待循环，应该每隔一段时间检查一下当前的线程状态，例如通过 Thread.currentThread().isInterrupted() 判断线程是否被打断，如果被打断则立即跳出循环。

下面是使用心跳检测机制的代码实例：

public class DistributedLock implements AutoCloseable {private static final String ROOT_NODE = "/businessCode"; // 业务根节点private static final long HEARTBEAT_RATE = 1000L; // 心跳检测频率，单位毫秒private final ZooKeeper zkClient; // ZooKeeper 客户端private final String lockName; // 分布式锁名称private volatile boolean locked = false; // 是否已经获取到了锁private Thread heartbeatThread; // 心跳线程private String currentLockNodePath; // 当前节点的 znode 路径public DistributedLock(String zkServers, String lockName) throws IOException {this.zkClient = new ZooKeeper(zkServers, 3000, event -> {});this.lockName = lockName;ensureRootNodeExists();}// 确保业务根节点存在private void ensureRootNodeExists() {try {if (zkClient.exists(ROOT_NODE, false) == null) {zkClient.create(ROOT_NODE, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}} catch (Exception e) {throw new RuntimeException("Failed to ensure root node exists: " + ROOT_NODE, e);}}// 获取锁public void lock() {try {// 创建当前节点currentLockNodePath = zkClient.create(ROOT_NODE + "/" + lockName + "_",null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);// 等待前一个节点释放锁while (!locked) {List<String> children = zkClient.getChildren(ROOT_NODE, false);String[] sortedNodes = children.stream().sorted().toArray(String[]::new);int currentNodeIndex = Arrays.binarySearch(sortedNodes, currentLockNodePath.substring(currentLockNodePath.lastIndexOf("/") + 1));if (currentNodeIndex == 0) {// 当前节点为业务节点下所有临时节点中的最小节点，获取锁locked = true;startHeartbeatThread(); // 启动心跳线程break;}// 等待前一个节点的删除事件通知String previousNodePath = ROOT_NODE + "/" + sortedNodes[currentNodeIndex - 1];CountDownLatch latch = new CountDownLatch(1);Stat stat = zkClient.exists(previousNodePath, event -> {if (event.getType() == Watcher.Event.EventType.NodeDeleted) {latch.countDown();}});if (stat != null) {latch.await();}}} catch (Exception e) {throw new RuntimeException("Failed to acquire lock: " + lockName, e);}}// 释放锁@Overridepublic void close() {try {System.out.println("Releasing lock: " + lockName);zkClient.delete(currentLockNodePath, -1);stopHeartbeatThread(); // 停止心跳线程zkClient.close();} catch (Exception e) {throw new RuntimeException("Failed to release lock: " + lockName, e);}}// 启动心跳线程private void startHeartbeatThread() {heartbeatThread = new Thread(() -> {while (true) {try {zkClient.setData(currentLockNodePath, null, -1);Thread.sleep(HEARTBEAT_RATE);} catch (Exception e) {System.out.println("Failed to send heartbeat signal: " + lockName);}}});heartbeatThread.start();}// 停止心跳线程private void stopHeartbeatThread() {if (heartbeatThread != null) {heartbeatThread.interrupt();heartbeatThread = null;}}
}

在该代码实例中，我们创建了一个布尔型变量 locked，表示当前是否已经获取到了锁。在获取锁的过程中，如果当前节点为业务节点下所有临时节点中的最小节点，则设置 locked 为 true，同时启动心跳线程。心跳线程每隔一段时间就向 ZooKeeper 发送一次空数据以保持会话，从而保证自己的临时节点不会过期。

在释放锁的过程中，停止心跳线程即可。需要注意的是，心跳线程的 while 循环不能被打断，因为一旦被打断，线程就会退出，从而不再发送心跳信号，导致临时节点过期。因此，在捕获 InterruptedException 异常时只是简单地输出日志，并继续下一轮循环。


http://www.ppmy.cn/news/47295.html

相关文章

科技云报到：存储开源，风雨飘摇下“披着羊皮的狼”？

科技云报到：存储开源，风雨飘摇下“披着羊皮的狼”？

科技云报道原创。 这些年开源界的风风雨雨&#xff0c;时不时撼动着人们的内心。 2022年&#xff0c;俄乌冲突导致全球最大的独立开源软件公司SUSE、美国开源软件巨头Redhat、主流开源容器引擎Docker&#xff0c;纷纷宣布停止与俄罗斯的合作。 而全球最大的开源及私有代码项目…
阅读更多...
socket 及 字节序转换（嵌入式学习）

socket 及 字节序转换（嵌入式学习）

socket 及 字节序转换 socket简介Socket为什么需要Socket&#xff1f;socket类型Socket通信模型 字节序主机字节序到网络字节序网络字节序到主机字节序IP地址转换 socket简介 1、1982 - Berkeley Software Distributions 操作系统引入了socket作为本地进程之间通信的接口 2、1…
阅读更多...
Linux 服务简单优化

Linux 服务简单优化

硬件优化 处理器&#xff1a;核心数、主频、制程工艺、线程数、缓存等 核心数&#xff1a;1、2、4、6、8、12、24、32等 主频&#xff1a;2.0GHz、2.3GHz等等 制程工艺&#xff1a;22nm、14nm、10nm等等 线程数&#xff1a;1、2 缓存&#xff1a;L1、L2、L3 建议&#xff1a;尽…
阅读更多...
Linux用户的分类与家目录，ls、pwd、cd、mkdir、touch、rmdir、rm指令与选项等

Linux用户的分类与家目录，ls、pwd、cd、mkdir、touch、rmdir、rm指令与选项等

Linux中用户的分类与用户的家目录 在Linux当中&#xff0c;用户的分类只分为两类&#xff0c;一类叫做超级用户root&#xff0c;还有就是其他也就是传说中的普通用户。我们刚刚登进去时&#xff0c;默认所处的目录是***/root或者/home/用户名***&#xff0c;比如说/root, /hom…
阅读更多...
知识库管理系统对于企业有哪些作用及优势？

知识库管理系统对于企业有哪些作用及优势？

知识库管理系统是一种通过集成多种技术手段&#xff0c;将企业内部知识进行收集、整理、存储、分析和共享的信息管理系统。知识库管理系统可以帮助企业管理和利用企业内部的知识&#xff0c;提高企业的创新能力和竞争力。 知识库管理系统的作用 1、促进企业内部知识的流通和共…
阅读更多...
无线传感器网络硬件设计简介

无线传感器网络硬件设计简介

无线传感器网络硬件设计简介 无线传感器网络因其巨大的应用前景越来越受到学术界和工业界的广泛关注。本文介绍了无线传感器网络节点的体系结构&#xff0c;分析比较了国内外当前典型的硬件平台&#xff0c;重点讨论了目前无线传感器网络节点常用的处理器、射频芯片、电源和传…
阅读更多...
Java IO流

Java IO流

Java IO流 IO就是Intput和Output也就是输入输出&#xff0c;将数据输入到计算机内存的过程叫做输入&#xff0c;从内存读取出来的数据叫做输出。Java的IO流总共有40多个类&#xff0c;他们都是由下个4个抽象类派生出来的&#xff0c;InputStream和Read和OutputStream和Writer。…
阅读更多...
网络拥塞控制，对越远的流量越宽容

网络拥塞控制，对越远的流量越宽容

考虑下面的网络传输场景&#xff1a; ​ S1&#xff0c;S2&#xff0c;S3&#xff0c;S4 向 D 方向发送&#xff0c;R4 发生拥塞。R4 必须丢弃一些数据进行疏导反馈拥塞信号&#xff0c;否则谁也过不去。 优先丢掉离得最近的 S4 的数据&#xff0c;其次依次丢 S3&#xff0c;S2…
阅读更多...
最新文章

Copyright @ 2022~2023