devbox
人工智能uu
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

基于zookeeper实现分布式锁_zk分布式锁

作者：人工智能uu | 2024-08-18 12:48:41

赞

踩

zk分布式锁

目录

zookeeper知识点复习

相关概念

java客户端操作

实现思路分析 

基本实现

初始化链接

代码落地 

优化：性能优化

 实现阻塞锁

监听实现阻塞锁

优化：可重入锁

zk分布式锁小结 

zookeeper知识点复习

Zookeeper（业界简称zk）是一种提供配置管理、分布式协同以及命名的中心化服务，这些提供的
功能都是分布式系统中非常底层且必不可少的基本功能，但是如果自己实现这些功能而且要达到高吞吐、低延迟同时还要保持一致性和可用性，实际上非常困难。因此zookeeper提供了这些功能，开发者在zookeeper之上构建自己的各种分布式系统。

相关概念

Zookeeper提供一个多层级的节点命名空间（节点称为znode），每个节点都用一个以斜杠（/）分隔的路径表示，而且每个节点都有父节点（根节点除外），非常类似于文件系统。并且每个节点都是唯一的。

znode节点有四种类型：

  • PERSISTENT：永久节点。客户端与zookeeper断开连接后，该节点依旧存在
  • EPHEMERAL：临时节点。客户端与zookeeper断开连接后，该节点被删除
  • PERSISTENT_SEQUENTIAL：永久节点、序列化。客户端与zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号
  • EPHEMERAL_SEQUENTIAL：临时节点、序列化。客户端与zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号

创建这四种节点：

 事件监听：在读取数据时，我们可以同时对节点设置事件监听，当节点数据或结构变化时，zookeeper会通知客户端。当前zookeeper有如下四种事件：
1. 节点创建
2. 节点删除
3. 节点数据修改
4. 子节点变更

java客户端操作

1. 引入依赖

  1. <dependency>
  2.     <groupId>org.apache.zookeeper</groupId>
  3.     <artifactId>zookeeper</artifactId>
  4.     <version>3.4.14</version>
  5. </dependency>

2. 常用api及其方法

初始化zookeeper客户端类，负责建立与zkServer的会话 

  1.        new ZooKeeper(connectString, 30000, new Watcher() {
  2.             @Override
  3.             public void process(WatchedEvent event) {
  4.                 System.out.println("获取链接成功！！");
  5.             }
  6.         });

创建一个节点，1-节点路径 2-节点内容 3-访问控制控制 4-节点类型 

  1.  
  2.  String fullPath = zooKeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
  3.                 CreateMode.PERSISTENT);

判断一个节点是否存在

  1. Stat stat = zooKeeper.exists(rootPath, false);
  2. if (stat != null) {...}

查询一个节点的内容 

  1.  Stat stat = new Stat();
  2.  byte[] data = zooKeeper.getData(path, false, stat);

更新一个节点 

zooKeeper.setData(rootPath, new byte[]{}, stat.getVersion() + 1);

删除一个节点 

 zooKeeper.delete(path, stat.getVersion());

查询一个节点的子节点列表 

List<String> children = zooKeeper.getChildren(rootPath, false);

关闭链接 

  1. if (zooKeeper != null) {
  2.        zooKeeper.close();
  3.    }

实现思路分析 

 分布式锁的步骤：
1. 获取锁：create一个节点
2. 删除锁：delete一个节点
3. 重试：没有获取到锁的请求重试
参照redis分布式锁的特点：
        1. 互斥 排他
        2. 防死锁：

        1. 可自动释放锁（临时节点） ：获得锁之后客户端所在机器宕机了，客户端没有主动删除子节点；如果创建的是永久的节点，那么这个锁永远不会释放，导致死锁；由于创建的是临时节点，客户端宕机后，过了一定时间zookeeper没有收到客户端的心跳包判断会话失效，将临时节点删除从而释放锁。
        2. 可重入锁：借助于ThreadLocal

3. 防误删：宕机自动释放临时节点，不需要设置过期时间，也就不存在误删问题。
4. 加锁/解锁要具备原子性
5. 单点问题：使用Zookeeper可以有效的解决单点问题，ZK一般是集群部署的。
6. 集群问题：zookeeper集群是强一致性的，只要集群中有半数以上的机器存活，就可以对外提供服务。

基本实现

实现思路：
1. 多个请求同时添加一个相同的临时节点，只有一个可以添加成功。添加成功的获取到锁
2. 执行业务逻辑
3. 完成业务流程后，删除节点释放锁。

初始化链接

由于zookeeper获取链接是一个耗时过程，这里可以在项目启动时，初始化链接，并且只初始化一次。借助于spring特性，代码实现如下：

  1. @Component
  2. public class zkClient {
  3.     private static final String connectString = "192.168.107.135";
  4.  
  5.     private static final String ROOT_PATH = "/distributed";
  6.  
  7.     private ZooKeeper zooKeeper;
  8.  
  9.     @PostConstruct
  10.     public void init() throws IOException {
  11.         this.zooKeeper = new ZooKeeper(connectString, 30000, new Watcher() {
  12.             @Override
  13.             public void process(WatchedEvent watchedEvent) {
  14.                 System.out.println("zookeeper 获取链接成功");
  15.             }
  16.         });
  17.         //创建分布式锁根节点
  18.         try {
  19.             if (this.zooKeeper.exists(ROOT_PATH, false) == null) {
  20.                 this.zooKeeper.create(ROOT_PATH, null,
  21.                         ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  22.             }
  23.         } catch (KeeperException e) {
  24.             e.printStackTrace();
  25.         } catch (InterruptedException e) {
  26.             e.printStackTrace();
  27.         }
  28.     }
  29.  
  30.     @PreDestroy
  31.     public void destroy() {
  32.         if (zooKeeper != null) {
  33.             try {
  34.                 zooKeeper.close();
  35.             } catch (InterruptedException e) {
  36.                 e.printStackTrace();
  37.             }
  38.         }
  39.     }
  40.     /**
  41.      * 初始化分布式对象方法
  42.      */
  43.     public ZkDistributedLock getZkDistributedLock(String lockname){
  44.         return new ZkDistributedLock(zooKeeper,lockname);
  45.     }
  46. }

代码落地 

  1. public class ZkDistributedLock {
  2.     public static final String ROOT_PATH = "/distribute";
  3.     private String path;
  4.     private ZooKeeper zooKeeper;
  5.  
  6.  
  7.     public ZkDistributedLock(ZooKeeper zooKeeper, String lockname) {
  8.         this.zooKeeper = zooKeeper;
  9.         this.path = ROOT_PATH + "/" + lockname;
  10.     }
  11.  
  12.     public void lock() {
  13.         try {
  14.             zooKeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
  15.         } catch (KeeperException e) {
  16.             e.printStackTrace();
  17.         } catch (InterruptedException e) {
  18.             e.printStackTrace();
  19.         }
  20.         try {
  21.             Thread.sleep(200);
  22.             lock();
  23.         } catch (InterruptedException e) {
  24.             e.printStackTrace();
  25.         }
  26.  
  27.     }
  28.  
  29.     public  void  unlock(){
  30.         try {
  31.             this.zooKeeper.delete(path,0);
  32.         } catch (InterruptedException e) {
  33.             e.printStackTrace();
  34.         } catch (KeeperException e) {
  35.             e.printStackTrace();
  36.         }
  37.  
  38.     }
  39. }

改造StockService的checkAndLock方法：

  1.     @Autowired
  2.     private zkClient client;
  3.     
  4.     public void checkAndLock() {
  5.         // 加锁，获取锁失败重试
  6.         ZkDistributedLock lock = this.client.getZkDistributedLock("lock");
  7.         lock.lock();
  8.         // 先查询库存是否充足
  9.         Stock stock = this.stockMapper.selectById(1L);
  10.         // 再减库存
  11.         if (stock != null && stock.getCount() > 0) {
  12.             stock.setCount(stock.getCount() - 1);
  13.             this.stockMapper.updateById(stock);
  14.         }
  15.         lock.unlock();
  16.     }

Jmeter压力测试：

 性能一般，mysql数据库的库存余量为0（注意：所有测试之前都要先修改库存量为5000）

基本实现存在的问题：
        1. 性能一般（比mysql略好）
        2. 不可重入
接下来首先来提高性能

优化：性能优化

基本实现中由于无限自旋影响性能：

试想：每个请求要想正常的执行完成，最终都是要创建节点，如果能够避免争抢必然可以提高性能。这里借助于zk的临时序列化节点，实现分布式锁： 

 实现阻塞锁

代码实现：

  1. public class ZkDistributedLock {
  2.     public static final String ROOT_PATH = "/distribute";
  3.     private String path;
  4.     private ZooKeeper zooKeeper;
  5.  
  6.  
  7.     public ZkDistributedLock(ZooKeeper zooKeeper, String lockname) {
  8.         this.zooKeeper = zooKeeper;
  9.         try {
  10.             this.path = zooKeeper.create(ROOT_PATH + "/" + lockname + "_",
  11.                     null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
  12.         } catch (KeeperException e) {
  13.             e.printStackTrace();
  14.         } catch (InterruptedException e) {
  15.             e.printStackTrace();
  16.         }
  17.     }
  18.  
  19.     public void lock() {
  20.         String preNode = getpreNode(path);
  21.         //如果该节点没有前一个节点，说明该节点是最小的节点
  22.         if (StringUtils.isEmpty(preNode)) {
  23.             return;
  24.         }
  25.         //重新检查是否获取到锁
  26.         try {
  27.             Thread.sleep(20);
  28.             lock();
  29.         } catch (InterruptedException e) {
  30.             e.printStackTrace();
  31.         }
  32.     }
  33.  
  34.     /**
  35.      * 获取指定节点的前节点
  36.      *
  37.      * @param path
  38.      * @return
  39.      */
  40.     private String getpreNode(String path) {
  41.         //获取当前节点的序列化序号
  42.         Long curSerial = Long.valueOf(StringUtil.substringAfter(path, '_'));
  43.         //获取根路径下的所有序列化子节点
  44.         try {
  45.             List<String> nodes = this.zooKeeper.getChildren(ROOT_PATH, false);
  46.             //判空处理
  47.             if (CollectionUtils.isEmpty(nodes)) {
  48.                 return null;
  49.             }
  50.             //获取前一个节点
  51.             Long flag = 0L;
  52.             String preNode = null;
  53.             for (String node : nodes) {
  54.                 //获取每个节点的序列化号
  55.                 Long serial = Long.valueOf(StringUtil.substringAfter(path, '_'));
  56.                 if (serial < curSerial && serial > flag) {
  57.                     flag = serial;
  58.                     preNode = node;
  59.                 }
  60.             }
  61.             return preNode;
  62.         } catch (KeeperException e) {
  63.             e.printStackTrace();
  64.         } catch (InterruptedException e) {
  65.             e.printStackTrace();
  66.         }
  67.         return null;
  68.     }
  69.  
  70.     public void unlock() {
  71.         try {
  72.             this.zooKeeper.delete(path, 0);
  73.         } catch (InterruptedException e) {
  74.             e.printStackTrace();
  75.         } catch (KeeperException e) {
  76.             e.printStackTrace();
  77.         }
  78.  
  79.     }
  80. }

主要修改了构造方法和lock方法：

 并添加了getPreNode获取前置节点的方法。

测试结果如下：

 性能反而更弱了。

原因：虽然不用反复争抢创建节点了，但是会自选判断自己是最小的节点，这个判断逻辑反而更复杂更 耗时。

解决方案：监听实现阻塞锁

监听实现阻塞锁

对于这个算法有个极大的优化点：假如当前有1000个节点在等待锁，如果获得锁的客户端释放锁时，这1000个客户端都会被唤醒，这种情况称为“羊群效应”；在这种羊群效应中，zookeeper需要通知1000个 客户端，这会阻塞其他的操作，最好的情况应该只唤醒新的最小节点对应的客户端。应该怎么做呢？在 设置事件监听时，每个客户端应该对刚好在它之前的子节点设置事件监听，例如子节点列表 为/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序号为1的客户端监听 序号为0的子节点删除消息，序号为2的监听序号为1的子节点删除消息。

所以调整后的分布式锁算法流程如下：

  • 客户端连接zookeeper，并在/lock下创建临时的且有序的子节点，第一个客户端对应的子节点 为/lock/lock-0000000000，第二个为/lock/lock-0000000001，以此类推；
  • 客户端获取/lock下的子节点列表，判断自己创建的子节点是否为当前子节点列表中序号最小的子 节点，如果是则认为获得锁，否则监听刚好在自己之前一位的子节点删除消息，获得子节点变更通 知后重复此步骤直至获得锁；
  • 执行业务代码；
  • 完成业务流程后，删除对应的子节点释放锁。

改造ZkDistributedLock的lock方法：

  1.     public void lock() {
  2.         String preNode = getpreNode(path);
  3.         //如果该节点没有前一个节点，说明该节点是最小的节点
  4.         if (StringUtils.isEmpty(preNode)) {
  5.             return;
  6.         } else {
  7.             CountDownLatch countDownLatch = new CountDownLatch(1);
  8.             try {
  9.                 if (this.zooKeeper.exists(ROOT_PATH + "/" + preNode, watchedEvent -> {
  10.                     countDownLatch.countDown();
  11.                 }) == null) {
  12.                     return;
  13.                 }
  14.                 countDownLatch.await();
  15.                 return;
  16.  
  17.             } catch (KeeperException e) {
  18.                 e.printStackTrace();
  19.             } catch (InterruptedException e) {
  20.                 e.printStackTrace();
  21.             }
  22.             try {
  23.                 Thread.sleep(200);
  24.             } catch (InterruptedException e) {
  25.                 e.printStackTrace();
  26.             }
  27.             lock();
  28.         }
  29.     }

压力测试效果如下：

 由此可见性能提高不少仅次于redis的分布式锁

优化：可重入锁

引入ThreadLocal线程局部变量保证zk分布式锁的可重入性。

在对应的线程的存储数据

  1. public class ZkDistributedLock {
  2.     public static final String ROOT_PATH = "/distribute";
  3.     private String path;
  4.     private ZooKeeper zooKeeper;
  5.     private static final ThreadLocal<Integer> THREAD_LOCAL = new ThreadLocal<>();
  6.  
  7.  
  8.     public ZkDistributedLock(ZooKeeper zooKeeper, String lockname) {
  9.         this.zooKeeper = zooKeeper;
  10.         try {
  11.             this.path = zooKeeper.create(ROOT_PATH + "/" + lockname + "_",
  12.                     null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
  13.         } catch (KeeperException e) {
  14.             e.printStackTrace();
  15.         } catch (InterruptedException e) {
  16.             e.printStackTrace();
  17.         }
  18.     }
  19.  
  20.     public void lock() {
  21.         Integer flag = THREAD_LOCAL.get();
  22.         if (flag != null && flag > 0) {
  23.             THREAD_LOCAL.set(flag + 1);
  24.             return;
  25.         }
  26.         String preNode = getpreNode(path);
  27.         //如果该节点没有前一个节点，说明该节点是最小的节点
  28.         if (StringUtils.isEmpty(preNode)) {
  29.             return;
  30.         } else {
  31.             CountDownLatch countDownLatch = new CountDownLatch(1);
  32.             try {
  33.                 if (this.zooKeeper.exists(ROOT_PATH + "/" + preNode, watchedEvent -> {
  34.                     countDownLatch.countDown();
  35.                 }) == null) {
  36.                     return;
  37.                 }
  38.                 countDownLatch.await();
  39.                 THREAD_LOCAL.set(1);
  40.                 return;
  41.  
  42.             } catch (KeeperException e) {
  43.                 e.printStackTrace();
  44.             } catch (InterruptedException e) {
  45.                 e.printStackTrace();
  46.             }
  47.             try {
  48.                 Thread.sleep(200);
  49.             } catch (InterruptedException e) {
  50.                 e.printStackTrace();
  51.             }
  52.             lock();
  53.         }
  54.     }
  55.  
  56.     /**
  57.      * 获取指定节点的前节点
  58.      *
  59.      * @param path
  60.      * @return
  61.      */
  62.     private String getpreNode(String path) {
  63.         //获取当前节点的序列化序号
  64.         Long curSerial = Long.valueOf(StringUtil.substringAfter(path, '_'));
  65.         //获取根路径下的所有序列化子节点
  66.         try {
  67.             List<String> nodes = this.zooKeeper.getChildren(ROOT_PATH, false);
  68.             //判空处理
  69.             if (CollectionUtils.isEmpty(nodes)) {
  70.                 return null;
  71.             }
  72.             //获取前一个节点
  73.             Long flag = 0L;
  74.             String preNode = null;
  75.             for (String node : nodes) {
  76.                 //获取每个节点的序列化号
  77.                 Long serial = Long.valueOf(StringUtil.substringAfter(path, '_'));
  78.                 if (serial < curSerial && serial > flag) {
  79.                     flag = serial;
  80.                     preNode = node;
  81.                 }
  82.             }
  83.             return preNode;
  84.         } catch (KeeperException e) {
  85.             e.printStackTrace();
  86.         } catch (InterruptedException e) {
  87.             e.printStackTrace();
  88.         }
  89.         return null;
  90.     }
  91.  
  92.     public void unlock() {
  93.         try {
  94.             THREAD_LOCAL.set(THREAD_LOCAL.get() - 1);
  95.             if (THREAD_LOCAL.get() == 0) {
  96.                 this.zooKeeper.delete(path, 0);
  97.                 THREAD_LOCAL.remove();
  98.             }
  99.  
  100.         } catch (InterruptedException e) {
  101.             e.printStackTrace();
  102.         } catch (KeeperException e) {
  103.             e.printStackTrace();
  104.         }
  105.  
  106.     }
  107. }

zk分布式锁小结 

参照redis分布式锁的特点：
1. 互斥 排他：zk节点的不可重复性，以及序列化节点的有序性
2. 防死锁：
        1. 可自动释放锁：临时节点
        2. 可重入锁：借助于ThreadLocal
3. 防误删：临时节点
4. 加锁/解锁要具备原子性
5. 单点问题：使用Zookeeper可以有效的解决单点问题，ZK一般是集群部署的。
6. 集群问题：zookeeper集群是强一致性的，只要集群中有半数以上的机器存活，就可以对外提供服务。
7. 公平锁：有序性节点

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/人工智能uu/article/detail/997520
推荐阅读
相关标签

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

  

闽ICP备14008679号