ZK（ZooKeeper）分布式锁实现

准备

本文会使用到 三台 独立服务器，可以自行提前搭建好。

不知道如何搭建的，可以看我之前 ZooKeeper集群 搭建：Zookeeper 集群部署的那些事儿

关于ZooKeeper 一些基础命令可以看这篇：Zookeeper入门看这篇就够了

前言

在平时我们对锁的使用，在针对单个服务，我们可以用 Java 自带的一些锁来实现，资源的顺序访问，但是随着业务的发展，现在基本上公司的服务都是多个，单纯的 Lock或者Synchronize 只能解决单个JVM线程的问题，那么针对于单个服务的 Java 的锁是无法满足我们业务的需要的，为了解决多个服务跨服务访问共享资源，于是就有了分布锁，分布式锁产生的原因就是集群。

正文

实现分布式锁的方式有哪些呢?

• 分布式锁的实现方式主要以(ZooKeeper、Reids、Mysql)这三种为主

今天我们主要讲解的是使用 ZooKeeper来实现分布式锁，ZooKeeper的应用场景主要包含这几个方面：

1. 服务注册与订阅(共用节点)
2. 分布式通知(监听ZNode)
3. 服务命令(ZNode特性)
4. 数据订阅、发布(Watcher)
5. 分布式锁(临时节点)

ZooKeeper实现分布式锁，主要是得益于ZooKeeper 保证了数据的强一致性，锁的服务可以分为两大类：

保持独占

所有试图来获取当前锁的客户端，最终有且只有一个能够成功得到当前锁的钥匙，通常我们会把 ZooKeeper 上的节点(ZNode)看做一把锁，通过 create临时节点的方式来实现，当多个客户端都去创建一把锁的时候，那么只有成功创建了那个客户端才能拥有这把锁

控制时序

所有试图获取锁的客户端，都是被顺序执行，只是会有一个序号(zxid)，我们会有一个节点，例如：/testLock，所有临时节点都在这个下面去创建，ZK的父节点(/testLock) 维持了一个序号，这个是ZK自带的属性，他保证了子节点创建的时序性，从而也形成了每个客户端的一个 全局时序

ZK锁机制

在实现ZooKeeper 分布式锁之前我们有必要了解一下，关于ZooKeeper分布式锁机制的实现流程和原理，不然各位宝贝，出去面试的时候怎么和面试官侃侃而谈~

临时顺序节点

基于ZooKeeper的临时顺序节点 ，ZooKeeper比较适合来实现分布式锁：

• 顺序发号器： ZooKeeper的每一个节点，都是自带顺序生成器：在每个节点下面创建临时节点，新的子节点后面，会添加一个次序编号，这个生成的编号，会在上一次的编号进行 +1 操作
• 有序递增： ZooKeeper节点有序递增，可以保证锁的公平性，我们只需要在一个持久父节点下，创建对应的临时顺序节点，每个线程在尝试占用锁之前，会调用watch，判断自己当前的序号是不是在当前父节点最小，如果是，那么获取锁
• Znode监听： 每个线程在抢占所之前，会创建属于当前线程的ZNode节点，在释放锁的时候，会删除创建的ZNode，当我们创建的序号不是最小的时候，会等待watch通知，也就是上一个ZNode的状态通知，当前一个ZNode删除的时候，会触发回调机制，告诉下一个ZNode，你可以获取锁开始工作了
• 临时节点自动删除： ZooKeeper还有一个好处，当我们客户端断开连接之后，我们出创建的临时节点会进行自动删除操作，所以我们在使用分布式锁的时候，一般都是会去创建临时节点，这样可以避免因为网络异常等原因，造成的死锁。
• 羊群效应： ZooKeeper节点的顺序访问性，后面监听前面的方式，可以有效的避免 羊群效应，什么是羊群效应：当某一个节点挂掉了，所有的节点都要去监听，然后做出回应，这样会给服务器带来比较大压力，如果有了临时顺序节点，当一个节点挂掉了，只有它后面的那一个节点才做出反应。

我们现在看一下下面一张图：

在上图中， ZooKeeper里面有一把锁节点 testLock，这个锁就是 ZooKeeper的一个节点，当两个客户端来获取这把锁的时候，会对 ZooKeeper进行加锁的请求，也就是我们所说的 临时顺序节点。

当我们在 /testLock目录下创建了一个顺序临时节点后，ZK会自动对这个临时节点维护 一个节点序号，并且这个节点是递增的，比如我们 clientA 创建了一个临时顺序节点，ZK内部会生成一个序号：/lock0000000001，那么 clientB 也生成了一个临时顺序节点，ZK会生成一个序号为 /lock0000000002，在这里数字都是依次递增的，从1开始递增，ZK内部会维护这个顺序。

下图所示：

这时候，ClientA会进行监听判断，在父节点下，我是不是最小的，如果是的话，那么俺就可以加锁了，因为我是最小的，其他的都比我大。我自己可以进行加锁，你已经是一个成熟的临时节点了，要学会自己加锁。咳，那么ZK是怎么进行判断的呢?宝贝，您往下看：

这个是 cleintA已经加锁完成了，这个时候 clientB也要过来加锁，那么他也要在 /testLock，创建一个属于自己的临时节点，那么这个时候他的序号就会变成 /lock0000000002，如下图所示：

这个时候就会出现我们前面所讲的，clientB 在加锁的时候会判断，自己是不是最小的，一看在当前父节点下不是最小的，啊~我还挺大的，还有比我小的!!!

加锁失败呀，咳咳，这个时候呢，clientB 就会去偷窥clientA，气氛逐渐暧昧起来，啊不是，是按照顺序去监听前一个节点(clientA)，是否完成工作了，如果完成了，clientB才可以进行加锁工作，宝贝，你往下看图片：

clientA 加锁成功后，会进行自己的业务处理，当 clientA 处理完工作后，说我完事了，下一个，那么 clientA 是怎么完事的呢，他多长时间?不是，具体流程是怎样的?小农你不对劲，说什么呢!!!真羞涩

上面我们不是说了，当 clientB 加锁失败后，会给前一个节点(clientA)加上一个监听，当clientA被删除以后，就表示有人释放了锁，这个时候就会通知 clientB重新去获取锁。

这个时候clientB重新获取锁的时候，发现自己就是当前父节点下面最小的那个，于是clientB就开始加锁，开始工作等一系列操作，当clientB 完事以后，释放锁，也说了一句，下一个。

如下图所示：

当然除了 clientA、clientB还有C\D\E等，这字母看着好奇怪又好熟悉，原理都是一样的，都是最小节点进行解锁，如果不是，监听前一个节点是否释放，如果释放了，再次尝试加锁。如果前一节节点释放了，自己就是最小了，就排到前面去了，有点类似于 银行取号 的操作。

代码实现

使用ZooKeeper 创建临时顺序节点来实现分布式锁，大体的流程就是 先创建一个持久父节点，在当前节点下，创建临时顺序节点，找出最小的序列号，获取分布式锁，程序业务完成之后释放锁，通知下一个节点进行操作，使用的是watch来监控节点的变化，然后依次下一个最小序列节点进行操作。

首先我们需要创建一个持久父类节点：我这里是 /mxn

WatchCallBack

1. importorg.apache.zookeeper.*;
2. importorg.apache.zookeeper.data.Stat;
3.  
4. importjava.util.Collections;
5. importjava.util.List;
6. importjava.util.concurrent.CountDownLatch;
7.  
8.  
9. /**
10. *@program:mxnzookeeper
11. *@ClassNameWatchCallBack
12. *@description:
13. *@author:微信搜索：牧小农
14. *@create:2021-10-2310:48
15. *@Version1.0
16. **/
17. publicclassWatchCallBackimplementsWatcher,AsyncCallback.StringCallback,AsyncCallback.Children2Callback,AsyncCallback.StatCallback{
18.  
19. ZooKeeperzk;
20. StringthreadName;
21. CountDownLatchcc=newCountDownLatch(1);
22. StringpathName;
23.  
24. publicStringgetPathName(){
25. returnpathName;
26. }
27.  
28. publicvoidsetPathName(StringpathName){
29. this.pathName=pathName;
30. }
31.  
32. publicStringgetThreadName(){
33. returnthreadName;
34. }
35.  
36. publicvoidsetThreadName(StringthreadName){
37. this.threadName=threadName;
38. }
39.  
40. publicZooKeepergetZk(){
41. returnzk;
42. }
43.  
44. publicvoidsetZk(ZooKeeperzk){
45. this.zk=zk;
46. }
47.  
48. /**@Author牧小农
49. *@Description//TODO尝试加锁方法
50. *@Date16:142021/10/24
51. *@Param
52. *@return
53. **/
54. publicvoidtryLock(){
55. try{
56.  
57. System.out.println(threadName+"开始创建。。。。");
58. //创建一个顺序临时节点
59. zk.create("/lock",threadName.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL,this,"abc");
60. //阻塞当前，监听前一个节点是否释放锁
61. cc.await();
62. }catch(InterruptedExceptione){
63. e.printStackTrace();
64. }
65. }
66.  
67. /**@Author牧小农
68. *@Description//TODO解锁方法
69. *@Date16:142021/10/24
70. *@Param
71. *@return
72. **/
73. publicvoidunLock(){
74. try{
75. //释放锁，删除临时节点
76. zk.delete(pathName,-1);
77. //结束工作
78. System.out.println(threadName+"结束工作了....");
79. }catch(InterruptedExceptione){
80. e.printStackTrace();
81. }catch(KeeperExceptione){
82. e.printStackTrace();
83. }
84. }
85.  
86.  
87. @Override
88. publicvoidprocess(WatchedEventevent){
89.  
90. //如果第一个节点释放了锁，那么第二个就会收到回调
91. //告诉它前一个节点释放了，你可以开始尝试获取锁
92. switch(event.getType()){
93. caseNone:
94. break;
95. caseNodeCreated:
96. break;
97. caseNodeDeleted:
98. //当前节点重新获取锁
99. zk.getChildren("/",false,this,"sdf");
100. break;
101. caseNodeDataChanged:
102. break;
103. caseNodeChildrenChanged:
104. break;
105. }
106.  
107. }
108.  
109. @Override
110. publicvoidprocessResult(intrc,Stringpath,Objectctx,Stringname){
111. if(name!=null){
112. System.out.println(threadName+"线程创建了一个节点为:"+name);
113. pathName=name;
114. //监听前一个节点
115. zk.getChildren("/",false,this,"sdf");
116. }
117.  
118. }
119.  
120. //getChildrencallback
121. @Override
122. publicvoidprocessResult(intrc,Stringpath,Objectctx,Listchildren,Statstat){
123.  
124. //节点按照编号，升序排列
125. Collections.sort(children);
126. //对节点进行截取例如/lock0000000022截取后就是lock0000000022
127. inti=children.indexOf(pathName.substring(1));
128.  
129.  
130. //是不是第一个，也就是说是不是最小的
131. if(i==0){
132. //是第一个
133. System.out.println(threadName+"现在我是最小的....");
134. try{
135. zk.setData("/",threadName.getBytes(),-1);
136. cc.countDown();
137.  
138. }catch(KeeperExceptione){
139. e.printStackTrace();
140. }catch(InterruptedExceptione){
141. e.printStackTrace();
142. }
143. }else{
144. //不是第一个
145. //监听前一个节点看它是不是完成了工作进行释放锁了
146. zk.exists("/"+children.get(i-1),this,this,"sdf");
147. }
148.  
149. }
150.  
151. @Override
152. publicvoidprocessResult(intrc,Stringpath,Objectctx,Statstat){
153. //判断是否失败exists
154. }
155. }

TestLock

1. importcom.mxn.zookeeper.config.ZKUtils;
2. importorg.apache.zookeeper.ZooKeeper;
3. importorg.junit.After;
4. importorg.junit.Before;
5. importorg.junit.Test;
6.  
7.  
8. /**
9. *@program:mxnzookeeper
10. *@ClassNameTestLock
11. *@description:
12. *@author:微信搜索：牧小农
13. *@create:2021-10-2310:45
14. *@Version1.0
15. **/
16. publicclassTestLock{
17.  
18.  
19. ZooKeeperzk;
20.  
21. @Before
22. publicvoidconn(){
23. zk=ZKUtils.getZK();
24. }
25.  
26. @After
27. publicvoidclose(){
28. try{
29. zk.close();
30. }catch(InterruptedExceptione){
31. e.printStackTrace();
32. }
33. }
34.  
35. @Test
36. publicvoidlock(){
37.  
38. //创建十个线程
39. for(inti=0;i<10;i++){
40. newThread(){
41. @Override
42. publicvoidrun(){
43. WatchCallBackwatchCallBack=newWatchCallBack();
44. watchCallBack.setZk(zk);
45. StringthreadName=Thread.currentThread().getName();
46. watchCallBack.setThreadName(threadName);
47. //线程进行抢锁操作
48. watchCallBack.tryLock();
49. try{
50. //进行业务逻辑处理
51. System.out.println(threadName+"开始处理业务逻辑了...");
52. Thread.sleep(200);
53. }catch(Exceptione){
54. e.printStackTrace();
55. }
56. //释放锁
57. watchCallBack.unLock();
58.  
59.  
60. }
61. }.start();
62. }
63.  
64.  
65. while(true){
66.  
67. }
68.  
69. }
70.  
71. }

运行结果:

1. Thread-1线程创建了一个节点为:/lock0000000112
2. Thread-5线程创建了一个节点为:/lock0000000113
3. Thread-2线程创建了一个节点为:/lock0000000114
4. Thread-6线程创建了一个节点为:/lock0000000115
5. Thread-9线程创建了一个节点为:/lock0000000116
6. Thread-4线程创建了一个节点为:/lock0000000117
7. Thread-7线程创建了一个节点为:/lock0000000118
8. Thread-3线程创建了一个节点为:/lock0000000119
9. Thread-8线程创建了一个节点为:/lock0000000120
10. Thread-0线程创建了一个节点为:/lock0000000121
11. Thread-1现在我是最小的....
12. Thread-1开始处理业务逻辑了...
13. Thread-1结束工作了....
14. Thread-5现在我是最小的....
15. Thread-5开始处理业务逻辑了...
16. Thread-5结束工作了....
17. Thread-2现在我是最小的....
18. Thread-2开始处理业务逻辑了...
19. Thread-2结束工作了....
20. Thread-6现在我是最小的....
21. Thread-6开始处理业务逻辑了...
22. Thread-6结束工作了....
23. Thread-9现在我是最小的....
24. Thread-9开始处理业务逻辑了...
25. Thread-9结束工作了....
26. Thread-4现在我是最小的....
27. Thread-4开始处理业务逻辑了...
28. Thread-4结束工作了....
29. Thread-7现在我是最小的....
30. Thread-7开始处理业务逻辑了...
31. Thread-7结束工作了....
32. Thread-3现在我是最小的....
33. Thread-3开始处理业务逻辑了...
34. Thread-3结束工作了....
35. Thread-8现在我是最小的....
36. Thread-8开始处理业务逻辑了...
37. Thread-8结束工作了....
38. Thread-0现在我是最小的....
39. Thread-0开始处理业务逻辑了...
40. Thread-0结束工作了....

总结

ZK分布式锁，能够有效的解决分布式、不可重入的问题，在上面的案例中我， 没有实现可重入锁，但是实现起来也不麻烦，只需要带上线程信息等唯一标识，判断一下就可以了

ZK实现分布式锁具有天然的优势，临时顺序节点，可以有效的避免死锁问题，让客户端断开，那么就会删除当前临时节点，让下一个节点进行工作。

如果文中有错误或者不了解的地方，欢迎留言，小农看见了会第一时间回复大家，大家加油

我是牧小农，一个卑微的打工人，如果觉得文中的内容对你有帮助，记得一键三连啊，你们的三连是小农最大的动力。

原文链接：https://mp.weixin.qq.com/s/qn8jaiE9ZortZ-wmDdceIg