devbox
喵喵爱编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Zookeeper——分布式锁的概念理解 & 应用举例_分布式锁的理解

作者：喵喵爱编程 | 2024-08-18 12:49:16

赞

踩

分布式锁的理解

文章目录：

1.前言

2.原生Zookeeper实现分布式锁 

3.Curator框架实现分布式锁案例

4.Zookeeper常见面试题

1.前言

什么叫做分布式锁呢？

比如说"进程 1"在使用该资源的时候，会先去获得锁，"进程 1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程 1"用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

2.原生Zookeeper实现分布式锁 

代码中的注释我已经写的很详细了。

这其中用到了JUC中的CountDownLatch，可以参考：https://blog.csdn.net/weixin_43823808/article/details/120799251

  1. package com.szh.case2;
  2.  
  3. import org.apache.zookeeper.*;
  4. import org.apache.zookeeper.data.Stat;
  5.  
  6. import java.io.IOException;
  7. import java.util.Collections;
  8. import java.util.List;
  9. import java.util.Objects;
  10. import java.util.concurrent.CountDownLatch;
  11.  
  12. /**
  13.  *
  14.  */
  15. public class DistributedZkLock {
  16.  
  17.     private final String connectString = "192.168.40.130:2181";
  18.     private final int sessionTimeout = 30000;
  19.     private final ZooKeeper zk;
  20.  
  21.     private CountDownLatch connectLatch = new CountDownLatch(1);
  22.     private CountDownLatch waitLatch = new CountDownLatch(1);
  23.  
  24.     private String currentNode;
  25.     private String waitPath;
  26.  
  27.     public DistributedZkLock() throws IOException, InterruptedException, KeeperException {
  28.         //获取zk连接
  29.         zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
  30.             @Override
  31.             public void process(WatchedEvent watchedEvent) {
  32.                 //如果连接上zk，则connectLatch可以释放
  33.                 if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
  34.                     connectLatch.countDown();
  35.                 }
  36.                 //如果监听的节点已被删除，同时当前监听的节点路径与即将被监听节点的前一个节点路径相同，则waitLatch可以释放
  37.                 if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
  38.                     waitLatch.countDown();
  39.                 }
  40.             }
  41.         });
  42.         //等待zk连接成功之后，程序则继续往下走，其他线程进入等待连接的状态
  43.         connectLatch.await();
  44.         //判断根节点 /locks 是否存在
  45.         Stat stat = zk.exists("/locks", false);
  46.         //如果根节点 /locks 不存在，则立马创建
  47.         if (Objects.isNull(stat)) {
  48.             //此 /locks 节点默认所有人均可访问，而且是永久性节点
  49.             zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  50.         }
  51.     }
  52.  
  53.     //加锁
  54.     public void zkLock() {
  55.         try {
  56.             //所谓加锁，就是在根节点/locks下创建对应的临时、带序号的节点
  57.             currentNode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
  58.             //睡一会，让结果更清晰
  59.             Thread.sleep(100);
  60.             //判断创建的节点是否是序号最小的节点，如果是，则获取到锁；如果不是，则监听它序号的前一个节点
  61.             List<String> children = zk.getChildren("/locks", false);
  62.             if (children.size() == 1) { //只有一个节点，则直接获取锁
  63.                 return;
  64.             } else { //如果有多个节点，则需要判断谁的序号最小
  65.                 //先对获取的节点的list集合排序，确保从小到大的顺序
  66.                 Collections.sort(children);
  67.                 //获取节点名称  seq-00000000
  68.                 String thisNode = currentNode.substring("/locks/".length());
  69.                 //通过节点名称获取到它在list集合中的位置
  70.                 int index = children.indexOf(thisNode);
  71.                 if (index == -1) { //没数据，无意义
  72.                     System.out.println("数据异常....");
  73.                 } else if (index == 0) { //说明此节点处于第一个位置，可以获取锁
  74.                     return;
  75.                 } else { //非第一个位置，需要监听前一个节点的变化
  76.                     //获取该节点序号的前一个节点
  77.                     waitPath = "/locks/" + children.get(index - 1);
  78.                     //监听，回调Watch的process方法
  79.                     zk.getData(waitPath, true, new Stat());
  80.                     //其他线程进入等待锁的状态
  81.                     waitLatch.await();
  82.                     return;
  83.                 }
  84.             }
  85.         } catch (KeeperException e) {
  86.             e.printStackTrace();
  87.         } catch (InterruptedException e) {
  88.             e.printStackTrace();
  89.         }
  90.     }
  91.  
  92.     //解锁
  93.     public void zkUnLock() {
  94.         //删除节点即解锁
  95.         try {
  96.             zk.delete(this.currentNode, -1);
  97.         } catch (InterruptedException e) {
  98.             e.printStackTrace();
  99.         } catch (KeeperException e) {
  100.             e.printStackTrace();
  101.         }
  102.     }
  103. }

下面是针对上面的一些方法的测试。 

  1. package com.szh.case2;
  2.  
  3. import org.apache.zookeeper.KeeperException;
  4.  
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeUnit;
  7.  
  8. /**
  9.  *
  10.  */
  11. public class DistributedZkLockTest {
  12.  
  13.     public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
  14.         //创建分布式锁1
  15.         final DistributedZkLock lock1 = new DistributedZkLock();
  16.         //创建分布式锁2
  17.         final DistributedZkLock lock2 = new DistributedZkLock();
  18.  
  19.         //如下创建两个线程，模拟获取分布式锁的过程
  20.         new Thread(new Runnable() {
  21.             @Override
  22.             public void run() {
  23.                 try {
  24.                     lock1.zkLock();
  25.                     System.out.println(Thread.currentThread().getName() + " 已启动，获取到锁....");
  26.                     TimeUnit.MILLISECONDS.sleep(3000);
  27.                     lock1.zkUnLock();
  28.                     System.out.println(Thread.currentThread().getName() + "已释放锁....");
  29.                 } catch (InterruptedException e) {
  30.                     e.printStackTrace();
  31.                 }
  32.             }
  33.         }).start();
  34.  
  35.         new Thread(new Runnable() {
  36.             @Override
  37.             public void run() {
  38.                 try {
  39.                     lock2.zkLock();
  40.                     System.out.println(Thread.currentThread().getName() + " 已启动，获取到锁....");
  41.                     TimeUnit.MILLISECONDS.sleep(3000);
  42.                     lock2.zkUnLock();
  43.                     System.out.println(Thread.currentThread().getName() + "已释放锁....");
  44.                 } catch (InterruptedException e) {
  45.                     e.printStackTrace();
  46.                 }
  47.             }
  48.         }).start();
  49.     }
  50. }

那么上面是我们自己手写的加锁、解锁的一些方法，其中也存在着很多问题。

（1）会话连接是异步的，需要自己去处理。比如使用 CountDownLatch
（2）Watch 需要重复注册，不然就不能生效
（3）开发的复杂性还是比较高的
（4）不支持多节点删除和创建。需要自己去递归
所以就引出了下面的案例：
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/喵喵爱编程/article/detail/997522
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。