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

zookeeper实用（2）：实现分布式锁

作者：Li_阴宅 | 2024-08-11 19:40:02

赞

踩

zookeeper实用（2）：实现分布式锁

场景：

分布式锁主要用于在分布式环境中保护跨进程、跨主机、跨网络的共享资源实现互斥访问，以达到保证数据的一致性。

在分布式环境下分布式锁的使用场景，比如秒杀情况，分布式任务，限制访问（同一时间只能发起一次）等。在单机环境中不推荐使用分布式锁，因为分布式锁使用会有一些性能开销的。

分布式锁的实现方式很多，比如DB，redis，zookeeper或者自己实现锁服务。

测试代码：

  1. package com.example.demo.util;
  2.  
  3. import org.I0Itec.zkclient.ZkClient;
  4.  
  5. import java.util.Comparator;
  6. import java.util.List;
  7. import java.util.concurrent.CountDownLatch;
  8.  
  9. public final class DLock
  10. {
  11.     private ZkClient zkClient;
  12.     private String lockName;
  13.     private String thisReadLock;
  14.     private String thisWriteLock;
  15.  
  16.     /**
  17.      * 分布式锁连接zookeeper 以及 初始化锁主节点
  18.      *
  19.      * @param hostUrl  zookeeper 连接url
  20.      * @param lockName 锁主节点
  21.      */
  22.     public void connect(String hostUrl , String lockName)
  23.     {
  24.         this.lockName = lockName;
  25.         zkClient = new ZkClient(hostUrl);
  26.         if (!zkClient.exists(lockName))
  27.             zkClient.createPersistent(lockName);
  28.     }
  29.  
  30.     /**
  31.      * 获取读锁
  32.      */
  33.     public void lockRead()
  34.     {
  35.         CountDownLatch readLatch = new CountDownLatch(1);
  36.         // 创建此临时节点， 获取带有顺序号的完整节点
  37.         String thisLockNodeBuilder = lockName + "/" + LockType.READ + "-";
  38.         thisReadLock = zkClient.createEphemeralSequential(thisLockNodeBuilder , "");
  39.  
  40.         // 找到此读锁前一个写锁。 找到节点的子节点
  41.         List<String> tmp_nodes = zkClient.getChildren(lockName);
  42.         sortNodes(tmp_nodes);
  43.         tmp_nodes.forEach(System.out::println);
  44.  
  45.         int tmp_index = 0;
  46.         // 倒序循环，直到读锁前面所有的写锁释放，读锁才执行
  47.         for (int i = tmp_nodes.size() - 1; i >= 0; i--)
  48.         {
  49.             if (thisReadLock.equals(lockName + "/" + tmp_nodes.get(i)))
  50.             {
  51.                 tmp_index = i;
  52.             } else if (i < tmp_index && tmp_nodes.get(i).split("-")[0].equals(LockType.WRITE.toString()))
  53.             {
  54.                 // 找到当前读锁之前的一个写锁
  55.                 // 先监听此写锁（监听该写锁是否释放），再阻塞当前读锁
  56.                 zkClient.subscribeChildChanges(lockName + "/" + tmp_nodes.get(i) , (parentPath , currentChilds) -> readLatch.countDown());
  57.  
  58.                 try
  59.                 {
  60.                     readLatch.await();
  61.                 } catch (InterruptedException e)
  62.                 {
  63.                     e.printStackTrace();
  64.                 }
  65.                 break;
  66.             }
  67.         }
  68.     }
  69.  
  70.     /**
  71.      * 释放读锁
  72.      */
  73.     public void unLockRead()
  74.     {
  75.         if (this.thisReadLock != null)
  76.         {
  77.             zkClient.delete(thisReadLock);
  78.             thisReadLock = null;
  79.         }
  80.     }
  81.  
  82.     /**
  83.      * 获取写锁
  84.      */
  85.     public void lockWrite()
  86.     {
  87.         CountDownLatch writeLatch = new CountDownLatch(1);
  88.         // 创建此临时节点， 获取带有顺序号的完整节点
  89.         String thisLockNodeBuilder = lockName + "/" + LockType.WRITE + "-";
  90.         thisWriteLock = zkClient.createEphemeralSequential(thisLockNodeBuilder , "");
  91.  
  92.         List<String> tmp_nodes = zkClient.getChildren(lockName);
  93.         sortNodes(tmp_nodes);
  94.         // 倒序循环，直到拿到最前面的写锁，它释放了，后面的写锁才能执行
  95.         for (int i = tmp_nodes.size() - 1; i >= 0; i--)
  96.         {
  97.             if (thisWriteLock.equals(lockName + "/" + tmp_nodes.get(i)))
  98.             {
  99.                 // 在锁列表中找到此写锁
  100.                 if (i > 0)
  101.                 {
  102.                     // 如果此写锁前面还有锁
  103.                     // 监听前面的锁(监听前面的锁是否释放)， 然后阻塞当前写锁获取
  104.                     zkClient.subscribeChildChanges(lockName + "/" + tmp_nodes.get(i - 1) , (parentPath , currentChilds) -> writeLatch.countDown());
  105.  
  106.                     try
  107.                     {
  108.                         // 阻塞当前写锁获取
  109.                         writeLatch.await();
  110.                     } catch (InterruptedException e)
  111.                     {
  112.                         e.printStackTrace();
  113.                     }
  114.                     break;
  115.                 }
  116.             }
  117.         }
  118.     }
  119.  
  120.     /**
  121.      * 释放写锁
  122.      */
  123.     public void unLockWrite()
  124.     {
  125.         if (thisWriteLock != null)
  126.         {
  127.             zkClient.delete(thisWriteLock);
  128.             thisWriteLock = null;
  129.         }
  130.     }
  131.  
  132.     /**
  133.      * 重新对节点进行排序，按照顺序号排序
  134.      *
  135.      * @param nodes 临时节点
  136.      */
  137.     private void sortNodes(List<String> nodes)
  138.     {
  139.         nodes.sort(Comparator.comparing(o -> o.split("-")[1]));
  140.     }
  141.  
  142.     /**
  143.      * 锁类型枚举
  144.      */
  145.     private enum LockType
  146.     {
  147.         READ,
  148.         WRITE;
  149.     }
  150. }

模拟多客户端并发请求：

  1. import com.example.demo.util.DLock;
  2. import org.springframework.beans.factory.annotation.Value;
  3.  
  4. public class TestDLock {
  5.  
  6.     private static String hostUrl = "127.0.0.1:2181";
  7.  
  8.     public static void main(String[] args) {
  9.         dLockClient1();
  10.         dLockClient2();
  11.         dLockClient3();
  12.         dLockClient4();
  13.     }
  14.  
  15.     public static void dLockClient1(){
  16.         // 客户端1
  17.         DLock lock=new DLock();
  18.         lock.connect(hostUrl, "/lock");
  19.         lock.lockRead();
  20.         System.out.println("I am testNode 1");
  21.         try
  22.         {
  23.             System.out.println("I am TestNode 1");
  24.             System.out.println("睡眠10s 之后释放分布式读锁, 开始倒计时");
  25.             for (int i = 0; i < 10; i++)
  26.             {
  27.                 System.out.println(10-i);
  28.                 Thread.sleep(1000);
  29.             }
  30.         } catch (InterruptedException e)
  31.         {
  32.             e.printStackTrace();
  33.         }
  34.         lock.unLockRead();
  35.     }
  36.  
  37.     public static void dLockClient2(){
  38.         // 客户端2
  39.         DLock lock=new DLock();
  40.         lock.connect(hostUrl, "/lock");
  41.         lock.lockRead();
  42.         System.out.println("I am TestNode 2");
  43.         lock.unLockRead();
  44.     }
  45.  
  46.     public static void dLockClient3(){
  47.         // 客户端3
  48.         DLock lock=new DLock();
  49.         lock.connect(hostUrl, "/lock");
  50.         lock.lockWrite();
  51.         System.out.println("I am testNode 3");
  52.         System.out.println("睡眠10s 之后释放分布式写锁, 开始倒计时");
  53.         for (int i = 0; i < 10; i++)
  54.         {
  55.             System.out.println(10-i);
  56.             try
  57.             {
  58.                 Thread.sleep(1000);
  59.             } catch (InterruptedException e)
  60.             {
  61.                 e.printStackTrace();
  62.             }
  63.         }
  64.         lock.unLockWrite();
  65.     }
  66.  
  67.     public static void dLockClient4(){
  68.         // 客户端4
  69.         DLock lock=new DLock();
  70.         lock.connect(hostUrl, "/lock");
  71.         lock.lockRead();
  72.         System.out.println("I am TestNode 4");
  73.         lock.unLockRead();
  74.     }
  75. }

运行结果：

  1. READ-0000000000
  2. I am testNode 1
  3. I am TestNode 1
  4. 睡眠10s 之后释放分布式读锁, 开始倒计时
  5. 10
  6. 9
  7. 8
  8. 7
  9. 6
  10. 5
  11. 4
  12. 3
  13. 2
  14. 1
  15. 17:44:29.643 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: 0x100026588980000 after 12ms
  16. 17:44:29.804 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980000, packet:: clientPath:null serverPath:null finished:false header:: 5,2  replyHeader:: 5,30064771079,0  request:: '/lock/READ-0000000000,-1  response:: null
  17. 17:44:29.805 [main] DEBUG org.I0Itec.zkclient.ZkConnection - Creating new ZookKeeper instance to connect to 127.0.0.1:2181.
  18. 17:44:29.805 [main] INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@1e643faf
  19. 17:44:29.805 [ZkClient-EventThread-15-127.0.0.1:2181] INFO org.I0Itec.zkclient.ZkEventThread - Starting ZkClient event thread.
  20. 17:44:29.809 [main] DEBUG org.I0Itec.zkclient.ZkClient - Awaiting connection to Zookeeper server
  21. 17:44:29.809 [main] INFO org.I0Itec.zkclient.ZkClient - Waiting for keeper state SyncConnected
  22. 17:44:29.810 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
  23. 17:44:29.812 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
  24. 17:44:29.812 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Session establishment request sent on 127.0.0.1/127.0.0.1:2181
  25. 17:44:29.827 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x100026588980001, negotiated timeout = 30000
  26. 17:44:29.827 [main-EventThread] DEBUG org.I0Itec.zkclient.ZkClient - Received event: WatchedEvent state:SyncConnected type:None path:null
  27. 17:44:29.827 [main-EventThread] INFO org.I0Itec.zkclient.ZkClient - zookeeper state changed (SyncConnected)
  28. 17:44:29.828 [main-EventThread] DEBUG org.I0Itec.zkclient.ZkClient - Leaving process event
  29. 17:44:29.828 [main] DEBUG org.I0Itec.zkclient.ZkClient - State is SyncConnected
  30. 17:44:29.838 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980001, packet:: clientPath:null serverPath:null finished:false header:: 1,3  replyHeader:: 1,30064771080,0  request:: '/lock,F  response:: s{30064771077,30064771077,1607247859574,1607247859574,0,2,0,0,0,0,30064771079} 
  31. 17:44:29.847 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980001, packet:: clientPath:null serverPath:null finished:false header:: 2,1  replyHeader:: 2,30064771081,0  request:: '/lock/READ-,#ffffffacffffffed057400,v{s{31,s{'world,'anyone}}},3  response:: '/lock/READ-0000000001 
  32. 17:44:29.851 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980001, packet:: clientPath:null serverPath:null finished:false header:: 3,8  replyHeader:: 3,30064771081,0  request:: '/lock,F  response:: v{'READ-0000000001} 
  33. READ-0000000001
  34. I am TestNode 2
  35. 17:44:29.859 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980001, packet:: clientPath:null serverPath:null finished:false header:: 4,2  replyHeader:: 4,30064771082,0  request:: '/lock/READ-0000000001,-1  response:: null
  36. 17:44:29.860 [main] DEBUG org.I0Itec.zkclient.ZkConnection - Creating new ZookKeeper instance to connect to 127.0.0.1:2181.
  37. 17:44:29.860 [main] INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@6e8dacdf
  38. 17:44:29.860 [ZkClient-EventThread-18-127.0.0.1:2181] INFO org.I0Itec.zkclient.ZkEventThread - Starting ZkClient event thread.
  39. 17:44:29.862 [main] DEBUG org.I0Itec.zkclient.ZkClient - Awaiting connection to Zookeeper server
  40. 17:44:29.862 [main] INFO org.I0Itec.zkclient.ZkClient - Waiting for keeper state SyncConnected
  41. 17:44:29.862 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
  42. 17:44:29.863 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
  43. 17:44:29.863 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Session establishment request sent on 127.0.0.1/127.0.0.1:2181
  44. 17:44:29.870 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x100026588980002, negotiated timeout = 30000
  45. 17:44:29.870 [main-EventThread] DEBUG org.I0Itec.zkclient.ZkClient - Received event: WatchedEvent state:SyncConnected type:None path:null
  46. 17:44:29.870 [main-EventThread] INFO org.I0Itec.zkclient.ZkClient - zookeeper state changed (SyncConnected)
  47. 17:44:29.871 [main-EventThread] DEBUG org.I0Itec.zkclient.ZkClient - Leaving process event
  48. 17:44:29.871 [main] DEBUG org.I0Itec.zkclient.ZkClient - State is SyncConnected
  49. 17:44:29.875 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980002, packet:: clientPath:null serverPath:null finished:false header:: 1,3  replyHeader:: 1,30064771083,0  request:: '/lock,F  response:: s{30064771077,30064771077,1607247859574,1607247859574,0,4,0,0,0,0,30064771082} 
  50. 17:44:29.882 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980002, packet:: clientPath:null serverPath:null finished:false header:: 2,1  replyHeader:: 2,30064771084,0  request:: '/lock/WRITE-,#ffffffacffffffed057400,v{s{31,s{'world,'anyone}}},3  response:: '/lock/WRITE-0000000002 
  51. 17:44:29.885 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980002, packet:: clientPath:null serverPath:null finished:false header:: 3,8  replyHeader:: 3,30064771084,0  request:: '/lock,F  response:: v{'WRITE-0000000002} 
  52. I am testNode 3
  53. 睡眠10s 之后释放分布式写锁, 开始倒计时
  54. 10
  55. 9
  56. 8
  57. 7
  58. 6
  59. 5
  60. 4
  61. 3
  62. 2
  63. 1
  64. 17:44:39.790 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: 0x100026588980000 after 5ms
  65. 17:44:39.864 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: 0x100026588980001 after 3ms
  66. 17:44:39.898 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: 0x100026588980002 after 5ms
  67. 17:44:39.992 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980002, packet:: clientPath:null serverPath:null finished:false header:: 4,2  replyHeader:: 4,30064771085,0  request:: '/lock/WRITE-0000000002,-1  response:: null
  68. 17:44:39.993 [main] DEBUG org.I0Itec.zkclient.ZkConnection - Creating new ZookKeeper instance to connect to 127.0.0.1:2181.
  69. 17:44:39.994 [main] INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@7a79be86
  70. 17:44:39.994 [ZkClient-EventThread-21-127.0.0.1:2181] INFO org.I0Itec.zkclient.ZkEventThread - Starting ZkClient event thread.
  71. 17:44:40.002 [main] DEBUG org.I0Itec.zkclient.ZkClient - Awaiting connection to Zookeeper server
  72. 17:44:40.002 [main] INFO org.I0Itec.zkclient.ZkClient - Waiting for keeper state SyncConnected
  73. 17:44:40.003 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
  74. 17:44:40.005 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
  75. 17:44:40.005 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Session establishment request sent on 127.0.0.1/127.0.0.1:2181
  76. 17:44:40.021 [main-SendThread(127.0.0.1:2181)] INFO org.apache.zookeeper.ClientCnxn - Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x100026588980003, negotiated timeout = 30000
  77. 17:44:40.021 [main-EventThread] DEBUG org.I0Itec.zkclient.ZkClient - Received event: WatchedEvent state:SyncConnected type:None path:null
  78. 17:44:40.021 [main-EventThread] INFO org.I0Itec.zkclient.ZkClient - zookeeper state changed (SyncConnected)
  79. 17:44:40.021 [main-EventThread] DEBUG org.I0Itec.zkclient.ZkClient - Leaving process event
  80. 17:44:40.022 [main] DEBUG org.I0Itec.zkclient.ZkClient - State is SyncConnected
  81. 17:44:40.032 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980003, packet:: clientPath:null serverPath:null finished:false header:: 1,3  replyHeader:: 1,30064771086,0  request:: '/lock,F  response:: s{30064771077,30064771077,1607247859574,1607247859574,0,6,0,0,0,0,30064771085} 
  82. 17:44:40.042 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980003, packet:: clientPath:null serverPath:null finished:false header:: 2,1  replyHeader:: 2,30064771087,0  request:: '/lock/READ-,#ffffffacffffffed057400,v{s{31,s{'world,'anyone}}},3  response:: '/lock/READ-0000000003 
  83. 17:44:40.046 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980003, packet:: clientPath:null serverPath:null finished:false header:: 3,8  replyHeader:: 3,30064771087,0  request:: '/lock,F  response:: v{'READ-0000000003} 
  84. READ-0000000003
  85. I am TestNode 4
  86. 17:44:40.052 [main-SendThread(127.0.0.1:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100026588980003, packet:: clientPath:null serverPath:null finished:false header:: 4,2  replyHeader:: 4,30064771088,0  request:: '/lock/READ-0000000003,-1  response:: null
  88. Process finished with exit code 0

 

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

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

  

闽ICP备14008679号