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

分布式锁实现方案-基于zookeeper的分布式锁实现（原理与代码）

作者：小蓝xlanll | 2024-04-23 10:25:21

赞

踩

分布式锁实现方案-基于zookeeper的分布式锁实现（原理与代码）

目录

一、基于zookeeper的分布式锁

1.1 基于Zookeeper实现分布式锁的原理

1.1.1 分布式锁特性说明

1.1.1.1 特点分析

1.1.1.2 本质

1.1.2 Zookeeper 分布式锁实现原理

1.1.2.1 Zookeeper临时顺序节点特性

1.1.2.2 Zookeeper满足分布式锁基本要求

1.1.2.3 Watcher机制

1.1.2.3 总结

1.2 分布式锁流程说明

1.2.1 分布式锁流程图

1.2.2 流程说明

1.3 分布式锁代码实现

1.3.1 自己手写，实现Lock接口

1.3.1.1 分布式锁ZookeeperDistributedLock

1.3.1.2 模拟下单处理OrderServiceHandle

1.3.1.3 订单号生成类OrderCodeGenerator

1.3.1.4 分布式锁测试类TestZookeeperDistributedLock

1.3.1.5 测试效果

1.3.2 基于Apache Curator 框架调用

1.3.2.1 maven依赖

1.3.2.2 代码实现

1.3.2.2.1 分布式锁类CuratorDistributeLock

1.3.2.2.2 测试类TestCuratorDistributedLock

1.3.2.3 执行结果

一、基于zookeeper的分布式锁

1.1 基于Zookeeper实现分布式锁的原理

1.1.1 分布式锁特性说明

1.1.1.1 特点分析
  • 每次只能一个占用锁；
  • 可以重复进入锁；
  • 只有占用者才可以解锁；
  • 获取锁和释放锁都需要原子
  • 不能产生死锁
  • 尽量满足性能
1.1.1.2 本质

同步互斥，使得处理任务能够一个一个逐步的过临界资源。

1.1.2 Zookeeper 分布式锁实现原理

1.1.2.1 Zookeeper临时顺序节点特性

zookeeper中有一种临时顺序节点，它具有以下特征：

  • 时效性，当会话结束，节点将自动被删除
  • 顺序性，当多个应用向其注册顺序节点时，每个顺序号将只能被一个应用获取
1.1.2.2 Zookeeper满足分布式锁基本要求
  1. 因为顺序性，可以让最小顺序号的应用获取到锁，从而满足分布式锁的 每次只能一个占用锁，因为只有它一个获取到，所以可以实现 重复进入 ，只要设置标识即可。锁的释放，即删除应用在zookeeper上注册的节点，因为每个节点只被自己注册拥有，所以只有自己才能删除，这样就满足只有占用者才可以解锁
  2. zookeeper的序号分配是原子的，分配后即不会再改变，让最小序号者获取锁，所以获取锁是原子的
  3. 因为注册的是临时节点，在会话期间内有效，所以不会产生死锁
  4. zookeeper注册节点的性能能满足几千，而且支持集群，能够满足大部分情况下的性能
1.1.2.3 Watcher机制

Zookeeper 允许客户端向服务端的某个 Znode 注册一个 Watcher 监听，当服务端的一些指定事
件触发了这个 Watcher，服务端会向指定客户端发送一个事件通知来实现分布式的通知功能，然
后客户端根据 Watcher 通知状态和事件类型做出业务上的改变。

在实现分布式锁的时候，主要利用这个机制，实现释放锁的时候，通知等待锁的线程竞争锁。

1.1.2.3 总结

综上可知，Zookeeper其实是基于临时顺序节点特性实现的分布式锁。当然，还结合了他的Watcher机制，实现释放锁的时候，通知等待锁的线程去竞争锁。

1.2 分布式锁流程说明

1.2.1 分布式锁流程图

1.2.2 流程说明

  1. client判断/lock目录是否存在，如果不存在则向其注册/lock的持久节点
  2. client向/lock/目录下注册/lock/Node-前缀的临时顺序节点，并得到顺序号
  3. client获取/lock/目录下的所有临时顺序子节点
  4. client判断临时子节点序号中是否存在比自身的序号小的节点。如果不存在，则获取到锁；如果存在，则对象该临时节点做watch监控
  5. 获得锁的线程，执行业务逻辑，执行完之后，删除临时节点，完成锁的释放。
  6. 等待锁的线程如果收到监控的临时节点被删除的通知，则再重复4、5、6步骤，进入下一个获得锁、释放锁的循环。

1.3 分布式锁代码实现

1.3.1 自己手写，实现Lock接口

1.3.1.1 分布式锁ZookeeperDistributedLock
  1. package com.ningzhaosheng.distributelock.zookeeper;
  2.  
  3. import org.I0Itec.zkclient.IZkDataListener;
  4. import org.I0Itec.zkclient.ZkClient;
  5. import org.I0Itec.zkclient.serialize.SerializableSerializer;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8.  
  9. import java.util.Collections;
  10. import java.util.List;
  11. import java.util.concurrent.CountDownLatch;
  12. import java.util.concurrent.TimeUnit;
  13. import java.util.concurrent.locks.Condition;
  14. import java.util.concurrent.locks.Lock;
  15.  
  16.  
  17. /**
  18.  * @author ningzhaosheng
  19.  * @date 2024/4/17 18:13:38
  20.  * @description 基于zookeeper实现的分布式锁
  21.  */
  22. public class ZookeeperDistributedLock implements Lock {
  23.  
  24.     private static Logger logger = LoggerFactory.getLogger(ZookeeperDistributedLock.class);
  25.  
  26.     // zookeeper 地址
  27.     private String ZOOKEEPER_IP_PORT = "192.168.152.130:2181";
  28.     // zookeeper 锁目录
  29.     private String LOCK_PATH = "/LOCK";
  30.  
  31.     // 创建 zookeeper客户端zkClient
  32.     private ZkClient client = null;
  33.  
  34.     private CountDownLatch cdl;
  35.  
  36.     // 当前请求的节点前一个节点
  37.     private String beforePath;
  38.     // 当前请求的节点
  39.     private String currentPath;
  40.  
  41.  
  42.     /**
  43.      * 初始化客户端和创建LOCK目录
  44.      *
  45.      * @param ZOOKEEPER_IP_PORT
  46.      * @param LOCK_PATH
  47.      */
  48.     public ZookeeperDistributedLock(String ZOOKEEPER_IP_PORT, String LOCK_PATH) {
  49.         this.ZOOKEEPER_IP_PORT = ZOOKEEPER_IP_PORT;
  50.         this.LOCK_PATH = LOCK_PATH;
  51.         client = new ZkClient(ZOOKEEPER_IP_PORT, 4000, 4000, new SerializableSerializer());
  52.         // 判断有没有LOCK目录，没有则创建
  53.         if (!this.client.exists(LOCK_PATH)) {
  54.             this.client.createPersistent(LOCK_PATH);
  55.         }
  56.     }
  57.  
  58.     @Override
  59.     public void lock() {
  60.         if (!tryLock()) {
  61.             //对次小节点进行监听
  62.             waitForLock();
  63.             lock();
  64.         } else {
  65.             logger.info(Thread.currentThread().getName() + " 获得分布式锁！");
  66.         }
  67.     }
  68.  
  69.     @Override
  70.     public void lockInterruptibly() throws InterruptedException {
  71.  
  72.     }
  73.  
  74.     @Override
  75.     public boolean tryLock() {
  76.         // 如果currentPath为空则为第一次尝试加锁，第一次加锁赋值currentPath
  77.         if (currentPath == null || currentPath.length() <= 0) {
  78.             // 创建一个临时顺序节点
  79.             currentPath = this.client.createEphemeralSequential(LOCK_PATH + '/', "lock");
  80.             System.out.println("---------------------------->" + currentPath);
  81.         }
  82.  
  83.         // 获取所有临时节点并排序，临时节点名称为自增长的字符串如：0000000400
  84.         List<String> childrens = this.client.getChildren(LOCK_PATH);
  85.         //由小到大排序所有子节点
  86.         Collections.sort(childrens);
  87.         //判断创建的子节点/LOCK/Node-n是否最小,即currentPath,如果当前节点等于childrens中的最小的一个就占用锁
  88.         if (currentPath.equals(LOCK_PATH + '/' + childrens.get(0))) {
  89.             return true;
  90.         }
  91.         //找出比创建的临时顺序节子节点/LOCK/Node-n次小的节点,并赋值给beforePath
  92.         else {
  93.             int wz = Collections.binarySearch(childrens, currentPath.substring(6));
  94.             beforePath = LOCK_PATH + '/' + childrens.get(wz - 1);
  95.         }
  96.  
  97.         return false;
  98.     }
  99.  
  100.     @Override
  101.     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  102.         return false;
  103.     }
  104.  
  105.     //等待锁,对次小节点进行监听
  106.     private void waitForLock() {
  107.         IZkDataListener listener = new IZkDataListener() {
  108.             public void handleDataDeleted(String dataPath) throws Exception {
  109.                 logger.info(Thread.currentThread().getName() + ":捕获到DataDelete事件！---------------------------");
  110.                 if (cdl != null) {
  111.                     cdl.countDown();
  112.                 }
  113.             }
  114.  
  115.             public void handleDataChange(String dataPath, Object data) throws Exception {
  116.  
  117.             }
  118.         };
  119.  
  120.         // 对次小节点进行监听,即beforePath-给排在前面的的节点增加数据删除的watcher
  121.         this.client.subscribeDataChanges(beforePath, listener);
  122.         if (this.client.exists(beforePath)) {
  123.             cdl = new CountDownLatch(1);
  124.             try {
  125.                 cdl.await();
  126.             } catch (InterruptedException e) {
  127.                 e.printStackTrace();
  128.             }
  129.         }
  130.         this.client.unsubscribeDataChanges(beforePath, listener);
  131.     }
  132.  
  133.     @Override
  134.     public void unlock() {
  135.         // 删除当前临时节点
  136.         client.delete(currentPath);
  137.     }
  138.  
  139.     @Override
  140.     public Condition newCondition() {
  141.         return null;
  142.     }
  143.  
  144. }
  145.
1.3.1.2 模拟下单处理OrderServiceHandle
  1. package com.ningzhaosheng.distributelock.zookeeper;
  2.  
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5.  
  6. import java.util.concurrent.CountDownLatch;
  7. import java.util.concurrent.locks.Lock;
  8.  
  9. /**
  10.  * @author ningzhaosheng
  11.  * @date 2024/4/17 21:45:46
  12.  * @description 模拟订单处理
  13.  */
  14. public class OrderServiceHandle implements Runnable {
  15.     private static OrderCodeGenerator ong = new OrderCodeGenerator();
  16.  
  17.     private Logger logger = LoggerFactory.getLogger(OrderServiceHandle.class);
  18.  
  19.     // 按照线程数初始化倒计数器,倒计数器
  20.     private CountDownLatch cdl = null;
  21.  
  22.     private Lock lock = null;
  23.  
  24.     public OrderServiceHandle(CountDownLatch cdl, Lock lock) {
  25.         this.cdl = cdl;
  26.         this.lock = lock;
  27.     }
  28.  
  29.     // 创建订单
  30.     public void createOrder() {
  31.         String orderCode = null;
  32.  
  33.         //准备获取锁
  34.         lock.lock();
  35.         try {
  36.             // 获取订单编号
  37.             orderCode = ong.getOrderCode();
  38.         } catch (Exception e) {
  39.             // TODO: handle exception
  40.         } finally {
  41.             //完成业务逻辑以后释放锁
  42.             lock.unlock();
  43.         }
  44.  
  45.         // ……业务代码
  46.  
  47.         logger.info("insert into DB使用id：=======================>" + orderCode);
  48.     }
  49.  
  50.  
  51.     @Override
  52.     public void run() {
  53.         try {
  54.             // 等待其他线程初始化
  55.             cdl.await();
  56.         } catch (InterruptedException e) {
  57.             // TODO Auto-generated catch block
  58.             e.printStackTrace();
  59.         }
  60.         // 创建订单
  61.         createOrder();
  62.     }
  63. }
  64.
1.3.1.3 订单号生成类OrderCodeGenerator
  1. package com.ningzhaosheng.distributelock.zookeeper;
  2.  
  3. import java.text.SimpleDateFormat;
  4. import java.util.Date;
  5.  
  6. /**
  7.  * @author ningzhaosheng
  8.  * @date 2024/4/17 21:44:06
  9.  * @description 生成订单号
  10.  */
  11. public class OrderCodeGenerator {
  12.     // 自增长序列
  13.     private static int i = 0;
  14.  
  15.     // 按照“年-月-日-小时-分钟-秒-自增长序列”的规则生成订单编号
  16.     public String getOrderCode() {
  17.         Date now = new Date();
  18.         SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
  19.         return sdf.format(now) + ++i;
  20.     }
  21. }
  22.
1.3.1.4 分布式锁测试类TestZookeeperDistributedLock
  1. package com.ningzhaosheng.distributelock.zookeeper;
  2.  
  3. import java.util.concurrent.CountDownLatch;
  4. import java.util.concurrent.locks.Lock;
  5.  
  6. /**
  7.  * @author ningzhaosheng
  8.  * @date 2024/4/17 21:48:28
  9.  * @description zookeeper分布式锁测试类
  10.  */
  11. public class TestZookeeperDistributedLock {
  12.     public static void main(String[] args) {
  13.         // zookeeper 地址
  14.         String ZOOKEEPER_IP_PORT = "192.168.31.9:2181";
  15.         // zookeeper 锁目录
  16.         String LOCK_PATH = "/LOCK";
  17.         // 线程并发数
  18.         int NUM = 10;
  19.  
  20.         CountDownLatch cdl = new CountDownLatch(NUM);
  21.         for (int i = 1; i <= NUM; i++) {
  22.             // 按照线程数迭代实例化线程
  23.             Lock lock = new ZookeeperDistributedLock(ZOOKEEPER_IP_PORT, LOCK_PATH);
  24.             new Thread(new OrderServiceHandle(cdl, lock)).start();
  25.             // 创建一个线程，倒计数器减1
  26.             cdl.countDown();
  27.         }
  28.     }
  29. }
  30.
1.3.1.5 测试效果

从上图执行结果中可以看出，在多线程情况下，分布式锁获取和释放正常。

1.3.2 基于Apache Curator 框架调用

1.3.2.1 maven依赖
  1.         <dependency>
  2.             <groupId>org.apache.curator</groupId>
  3.             <artifactId>curator-framework</artifactId>
  4.             <version>5.2.0</version>
  5.         </dependency>
  6.  
  7.         <dependency>
  8.             <groupId>org.apache.curator</groupId>
  9.             <artifactId>curator-recipes</artifactId>
  10.             <version>5.2.0</version>
  11.         </dependency>
1.3.2.2 代码实现

这里模拟业务使用分布式锁，还是使用的OrderServiceHandle类，这里只给出分布式锁实现类和测试类，不再给出OrderServiceHandle代码，可以参考上一小节的OrderServiceHandle类。

1.3.2.2.1 分布式锁类CuratorDistributeLock
  1. package com.ningzhaosheng.distributelock.zookeeper.curator;
  2.  
  3. import org.apache.curator.framework.CuratorFramework;
  4. import org.apache.curator.framework.CuratorFrameworkFactory;
  5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
  6. import org.apache.curator.retry.ExponentialBackoffRetry;
  7.  
  8. import java.util.concurrent.TimeUnit;
  9. import java.util.concurrent.locks.Condition;
  10. import java.util.concurrent.locks.Lock;
  11.  
  12. /**
  13.  * @author ningzhaosheng
  14.  * @date 2024/4/17 22:03:45
  15.  * @description 实现Lock接口（其实可以不用，可以直接使用InterProcessMutex，这里是为了和jvm的Lock锁保持一致，所以做了一层封装）
  16.  */
  17. public class CuratorDistributeLock implements Lock {
  18.  
  19.     private CuratorFramework client;
  20.     private InterProcessMutex mutex;
  21.  
  22.     public CuratorDistributeLock(String connString, String lockPath) {
  23.         this(connString, lockPath, new ExponentialBackoffRetry(3000,5));
  24.     }
  25.  
  26.     public CuratorDistributeLock(String connString, String lockPath, ExponentialBackoffRetry retryPolicy) {
  27.         try {
  28.             client = CuratorFrameworkFactory.builder()
  29.                     .connectString(connString)
  30.                     .retryPolicy(retryPolicy)
  31.                     .build();
  32.             client.start();
  33.  
  34.             mutex = new InterProcessMutex(client, lockPath);
  35.         } catch (Exception e) {
  36.             e.printStackTrace();
  37.         }
  38.     }
  39.  
  40.     @Override
  41.     public void lock() {
  42.         try {
  43.             // 获取锁
  44.             mutex.acquire();
  45.         } catch (Exception e) {
  46.             e.printStackTrace();
  47.         }
  48.     }
  49.  
  50.     @Override
  51.     public void lockInterruptibly() throws InterruptedException {
  52.  
  53.     }
  54.  
  55.     @Override
  56.     public boolean tryLock() {
  57.         return false;
  58.     }
  59.  
  60.     @Override
  61.     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  62.         return false;
  63.     }
  64.  
  65.     @Override
  66.     public void unlock() {
  67.         try {
  68.             // 释放锁
  69.             mutex.release();
  70.         } catch (Exception e) {
  71.             e.printStackTrace();
  72.         }
  73.     }
  74.  
  75.     @Override
  76.     public Condition newCondition() {
  77.         return null;
  78.     }
  79. }
  80.
1.3.2.2.2 测试类TestCuratorDistributedLock
  1. package com.ningzhaosheng.distributelock.zookeeper.curator;
  2.  
  3. import com.ningzhaosheng.distributelock.zookeeper.OrderServiceHandle;
  4.  
  5. import java.util.concurrent.CountDownLatch;
  6.  
  7. /**
  8.  * @author ningzhaosheng
  9.  * @date 2024/4/17 21:54:33
  10.  * @description 基于 apache curator分布式锁测试类
  11.  */
  12. public class TestCuratorDistributedLock {
  13.     private static final String ZK_ADDRESS = "192.168.31.9:2181";
  14.     private static final String LOCK_PATH = "/distributed_lock";
  15.  
  16.     public static void main(String[] args) {
  17.  
  18.         int NUM = 10;
  19.         CountDownLatch cdl = new CountDownLatch(NUM);
  20.         for (int i = 1; i <= NUM; i++) {
  21.             // 按照线程数迭代实例化线程
  22.             /** 创建CuratorDistributeLock
  23.              * 其实可以不用，可以直接使用InterProcessMutex，这里是为了和jvm的Lock锁保持一致，所以做了一层封装
  24.              */
  25.             CuratorDistributeLock curatorDistributeLock = new CuratorDistributeLock(ZK_ADDRESS,LOCK_PATH);
  26.             new Thread(new OrderServiceHandle(cdl, curatorDistributeLock)).start();
  27.             // 创建一个线程，倒计数器减1
  28.             cdl.countDown();
  29.         }
  30.     }
  31. }
  32.
1.3.2.3 执行结果

从执行结果可以看出，基于apche curator框架实现zookeeper锁，它也是按照临时顺序节点的顺序获取锁的，每次获得锁的节点都是最小顺序节点，然后等待锁的线程，会基于watcher机制，每次给最小临时顺序节点加回调，监听节点的变更（即释放锁的线程会删除节点），然后再重新判断最小临时顺序节点，最小的获得锁执行，依次循环完成。

好了，本次内容就分享到这，欢迎关注本博主。如果有帮助到大家，欢迎大家点赞+关注+收藏，有疑问也欢迎大家评论留言！

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

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

  

闽ICP备14008679号