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

基于ZooKeeper的分布式锁

作者：很楠不爱3 | 2024-05-19 14:54:59

赞

踩

基于ZooKeeper的分布式锁

1.为什么要使用分布式锁？

       在我们以前的单进程多线程里面，会用到多线程来提高工作效率。在我们目前计算机多核心的背景下，每个CPU核心能够同时运行一个线程，在同一时刻就会出现多线程并行问题。如果多线程并行去访问某一个共享资源的话，这个时候就有可能会造成共享资源的线程安全问题。在多线程中，我们可以使用Synchronized或者Lock来解决单进程多线程里面的线程安全问题。

       在分布式架构中，整个架构的每个节点都是一个独立的进程，每个进程都能独立的去完成一个操作。例如：电商平台中商品库存，100个库存，1000人抢购的情况。并行操作下，如何对库存进行处理，以防卖多或卖少问题，这种情况，在分布式环境下，通过synchronized或者Lock是无法解决的。

       为了解决这类问题，我们需要通过一些机制来实现分布式锁。我们可以通过Redis、数据库、ZooKeeper，这些都能用来实现分布式锁，我们只需要依赖一个第三方中间件，便能够解决这一类问题。此处我们学习如何使用ZooKeeper原生API来实现分布式锁。

2.ZooKeeper如何实现分布式锁？

        通过一个情景，我们来介绍ZooKeeper如何实现分布式锁。

        现在有3个客户端，分别是ClientA、ClientB、ClientC，都去访问一个共享资源，调用ZooKeeper拿到一个锁。一共有两种方式：

        ①利用ZooKeeper同节点路径唯一性(不建议使用)

         在ZooKeeper的一个/Locks节点下，去创建一个子节点/lock。3个客户端同时访问一个共享资源，那么每个客户端都会去ZooKeeper上创建一个/lock子节点。由于ZooKeeper节点特性，在同一级节点中路径的唯一性，导致3个客户端只有一个会创建成功。那么失败的客户端，都会去监听/lock这个子节点的变化，子节点发生变化时，都会去触发一个watch通知，其他客户端便会收到通知，再去争抢锁。

       这种实现非常简单，但是这种方式会产生一个问题，那就是惊群效应。那么什么是惊群效应？比如我们现在有100个节点，如果1个节点释放了锁，那么其他99个节点都会去争抢锁。一个触发机制让所有参与的客户端都去发起争抢，这就是惊群效应，也叫作羊群效应。

       这样一个惊群效应，会导致在短时间内有大量的事件变更，实际情况下是只会有一个客户端获得锁，所以这种情况下，在集群规模大或者访问节点数量多的情况下，不建议使用这种方式来实现分布式锁。

        ②利用ZooKeeper有序节点特性(推荐使用)

        每个客户端，在/Locks节点下分别创建一个带序号的临时节点，这样子我们就能够保证每一个客户端都能够去注册一个节点到ZooKeeper上去。现在客户端要去获得锁，只需要从/Locks这个节点下的所有子节点中，获取一个最小的节点获得锁。剩下的其他节点之间，做相互监听即可。那么什么是相互监听呢？

        现在有/001、/002、/003三个有序临时节点，现在/001节点获取锁，则对应的客户端A获得锁(节点/001是客户端A注册到zk的节点)，那么其他节点都会去监听这个节点的变化。如果这个节点发生了一个删除或者失效的变化，会产生一个watch监听，让每个节点只去监听比自己小一个值的节点即可（即：/003监听/002，/002监听/001，这就是相互监听。目前/001获得锁，如果/001节点失效，则/002监听到/001失效后，/002会获得锁）。如果这个节点发生变化，监听者能收到一个watch监听事件，再去判断当前节点是不是所有节点里面最小的节点。如果是最小节点，则让它获得锁；如果不是，继续等待，直到获得锁。

3.ZooKeeper代码实现分布式锁(利用ZK有序节点特性)

  1. package zookeeper;
  2.  
  3. import org.apache.zookeeper.*;
  4. import org.apache.zookeeper.data.Stat;
  5.  
  6. import java.io.IOException;
  7. import java.util.List;
  8. import java.util.SortedSet;
  9. import java.util.TreeSet;
  10. import java.util.concurrent.CountDownLatch;
  11. import java.util.concurrent.TimeUnit;
  12. import java.util.concurrent.locks.Condition;
  13. import java.util.concurrent.locks.Lock;
  14.  
  15. /**
  16.  * ZooKeeper分布式锁实现
  17.  * 实现Lock接口、Watch接口,重写接口方法
  18.  */
  19. public class DistributedLock implements Lock, Watcher {
  20.  
  21.     //定义ZooKeeper连接
  22.     private ZooKeeper zk = null;
  23.     //定义根节点(当前在那个根节点创建锁)
  24.     private String ROOT_LOCK = "/lock";
  25.     //定义一个等待的锁(等待前一个锁)
  26.     private String WAIT_LOCK;
  27.     //定义当前获得锁的节点
  28.     private String CURRENT_LOCK;
  29.  
  30.     //用来做锁控制
  31.     private CountDownLatch countDownLatch;
  32.  
  33.     //无参构造,用于初始化操作
  34.     public DistributedLock() {
  35.         try {
  36.             zk = new ZooKeeper("192.168.204.202:2181", 4000, this);
  37.             //判断根节点是否存在(watch为false,因为我不需要对当前事件再去监听)
  38.             Stat stat = zk.exists(ROOT_LOCK,false);
  39.             //如果不存在，创建持久化节点
  40.             if(stat == null){
  41.                 zk.create(ROOT_LOCK,"0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  42.             }
  43.  
  44.         } catch (IOException e) {
  45.             e.printStackTrace();
  46.         }catch (KeeperException e) {
  47.             e.printStackTrace();
  48.         } catch (InterruptedException e) {
  49.             e.printStackTrace();
  50.         }
  51.     }
  52.  
  53.  
  54.     @Override
  55.     public void lock() {
  56.         if(this.tryLock()){
  57.             System.out.println(Thread.currentThread().getName() + "-->" + CURRENT_LOCK + "获得锁成功");
  58.             return;
  59.         }
  60.         waitForLock(WAIT_LOCK);//没有获得锁，继续等待获得锁
  61.     }
  62.  
  63.     //持续阻塞去获得锁
  64.     private boolean waitForLock(String prev) {
  65.         try {
  66.             //监听当前节点的上一个节点
  67.             Stat stat = zk.exists(prev,true);
  68.             if(stat != null){
  69.                 System.out.println(Thread.currentThread().getName() + "-->等待锁" + prev + "释放");
  70.                 countDownLatch = new CountDownLatch(1);
  71.                 countDownLatch.await();
  72.                 System.out.println(Thread.currentThread().getName() + "-->获得锁成功");
  73.             }
  74.  
  75.         } catch (KeeperException e) {
  76.             e.printStackTrace();
  77.         } catch (InterruptedException e) {
  78.             e.printStackTrace();
  79.         }
  80.         return true;
  81.     }
  82.  
  83.     @Override
  84.     public void lockInterruptibly() throws InterruptedException {
  85.  
  86.     }
  87.  
  88.     //每次调用tryLock
  89.     @Override
  90.     public boolean tryLock() {
  91.         try {
  92.             //创建一个临时有序节点(设置给CURRENT_LOCK)
  93.             CURRENT_LOCK = zk.create(ROOT_LOCK+"/","0".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
  94.  
  95.             System.out.println(Thread.currentThread().getName() + "-->" + CURRENT_LOCK + ",尝试竞争锁");
  96.  
  97.             //获取当前的Children,判断哪个节点是最小的
  98.             List<String> childrens = zk.getChildren(ROOT_LOCK , false);
  99.             //对其进行排序
  100.             SortedSet<String> sortedSet = new TreeSet<String>();
  101.             //所有节点放入SortedSet进行排序
  102.             for(String children : childrens){
  103.                 sortedSet.add(ROOT_LOCK+"/"+children);
  104.             }
  105.             //得到最小的节点
  106.             String firstNode = sortedSet.first();
  107.             //判断当前节点是否是子节点中最小的(eg:如果不是，会返回比他小的节点)
  108.             SortedSet<String> lessThenMe = ((TreeSet<String>) sortedSet).headSet(CURRENT_LOCK);
  109.             //判断当前节点是否是最小节点(当前节点与子节点中最小节点比较，如果相等，则获得锁)
  110.             if(CURRENT_LOCK.equals(firstNode)){
  111.                 return true;
  112.             }
  113.             //没有比自己更小的节点
  114.             if(!lessThenMe.isEmpty()){
  115.                 WAIT_LOCK = lessThenMe.last();//获取比当前节点更小的最后一个节点，设置给WAIT_LOCK去等待
  116.             }
  117.  
  118.         } catch (KeeperException e) {
  119.             e.printStackTrace();
  120.         } catch (InterruptedException e) {
  121.             e.printStackTrace();
  122.         }
  123.         return false;
  124.     }
  125.  
  126.     @Override
  127.     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  128.         return false;
  129.     }
  130.  
  131.     //释放锁
  132.     @Override
  133.     public void unlock() {
  134.         try {
  135.             System.out.println(Thread.currentThread().getName()+"-->释放锁"+CURRENT_LOCK);
  136.             zk.delete(CURRENT_LOCK,-1);
  137.             CURRENT_LOCK = null;
  138.             zk.close();
  139.         } catch (InterruptedException e) {
  140.             e.printStackTrace();
  141.         } catch (KeeperException e) {
  142.             e.printStackTrace();
  143.         }
  144.     }
  145.  
  146.     @Override
  147.     public Condition newCondition() {
  148.         return null;
  149.     }
  150.  
  151.     //处理监听
  152.     @Override
  153.     public void process(WatchedEvent watchedEvent) {
  154.         if(this.countDownLatch != null){
  155.             this.countDownLatch.countDown();
  156.         }
  157.     }
  158. }

测试类

  1. /**
  2.  * 测试ZooKeeper分布式锁
  3.  */
  4. public class DistrutedLockDemo {
  5.     public static void main(String[] args) throws IOException {
  6.         CountDownLatch countDownLatch = new CountDownLatch(10);
  7.         for (int i = 0; i < 10 ; i++) {
  8.             new Thread(()->{
  9.                 try {
  10.                     countDownLatch.await();
  11.                     DistributedLock distributedLock = new DistributedLock();
  12.                     distributedLock.lock();//获得锁
  13.                 } catch (InterruptedException e) {
  14.                     e.printStackTrace();
  15.                 }
  16.             },"Thread-"+i).start();
  17.             countDownLatch.countDown();
  18.         }
  19.         System.in.read();
  20.     }
  21. }

结果：

      我们可以看到，刚开始Thread-8获得锁，对应的是/0000000070节点，当我们删除70节点，最小的子节点则为71，所以Thread-6获得锁，以此类推，便能够满足多进程分布式锁操作。如下图所示

缺点：

      基于ZooKeeper API实现的分布式锁，编写代码相对复杂繁琐，为了简化编写过程，请移步Curator分布式锁的实现。

附：基于Redis分布式锁的实现(Jedis)

博主写作不易，来个关注呗

求关注、求点赞，加个关注不迷路 ヾ(◍°∇°◍)ﾉﾞ

博主不能保证写的所有知识点都正确，但是能保证纯手敲，错误也请指出，望轻喷 Thanks♪(･ω･)ﾉ

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

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

  

闽ICP备14008679号