ZK（ZooKeeper）分布式锁实现
这时候，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
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
/**
@program: mxnzookeeper
@ClassName WatchCallBack
@description:
@author: 微信搜索：牧小农
@create: 2021-10-23 10:48
@Version 1.0
**/
public class WatchCallBack implements Watcher, AsyncCallback.StringCallback ,AsyncCallback.Children2Callback ,AsyncCallback.StatCallback {
ZooKeeper zk ;
String threadName;
CountDownLatch cc = new CountDownLatch(1);
String pathName;
public String getPathName() {
return pathName;
}
public void setPathName(String pathName) {
this.pathName = pathName;
}
public String getThreadName() {
return threadName;
}
public void setThreadName(String threadName) {
this.threadName = threadName;
}
public ZooKeeper getZk() {
return zk;
}
public void setZk(ZooKeeper zk) {
this.zk = zk;
}
/** @Author 牧小农
@Description //TODO 尝试加锁方法
@Date 16:14 2021/10/24
@Param
@return
**/
public void tryLock(){
try {
System.out.println(threadName + " 开始创建。。。。");
//创建一个顺序临时节点
zk.create("/lock",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,this,"abc");
//阻塞当前，监听前一个节点是否释放锁
cc.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/** @Author 牧小农
@Description //TODO 解锁方法
@Date 16:14 2021/10/24
@Param
@return
**/
public void unLock(){
try {
//释放锁，删除临时节点
zk.delete(pathName,-1);
//结束工作
System.out.println(threadName + " 结束工作了....");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
@Override
public void process(WatchedEvent event) {
//如果第一个节点释放了锁，那么第二个就会收到回调
//告诉它前一个节点释放了，你可以开始尝试获取锁
switch (event.getType()) {
case None:
break;
case NodeCreated:
break;
case NodeDeleted:
//当前节点重新获取锁
zk.getChildren("/",false,this ,"sdf");
break;
case NodeDataChanged:
break;
case NodeChildrenChanged:
break;
}
}
@Override
public void processResult(int rc, String path, Object ctx, String name) {
if(name != null ){
System.out.println(threadName +" 线程创建了一个节点为 : " + name );
pathName = name ;
//监听前一个节点
zk.getChildren("/",false,this ,"sdf");
}
}
//getChildren call back
@Override
public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
//节点按照编号，升序排列
Collections.sort(children);
//对节点进行截取例如 /lock0000000022 截取后就是 lock0000000022
int i = children.indexOf(pathName.substring(1));
//是不是第一个，也就是说是不是最小的
if(i == 0){
//是第一个
System.out.println(threadName +" 现在我是最小的....");
try {
zk.setData("/",threadName.getBytes(),-1);
cc.countDown();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}else{
//不是第一个
//监听前一个节点 看它是不是完成了工作进行释放锁了
zk.exists("/"+children.get(i-1),this,this,"sdf");
}
}
@Override
public void processResult(int rc, String path, Object ctx, Stat stat) {
//判断是否失败 exists
}
}
TestLock
import com.mxn.zookeeper.config.ZKUtils;
import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
/**
@program: mxnzookeeper
@ClassName TestLock
@description:
@author: 微信搜索：牧小农
@create: 2021-10-23 10:45
@Version 1.0
**/
public class TestLock {
ZooKeeper zk ;
@Before
public void conn (){
zk = ZKUtils.getZK();
}
@After
public void close (){
try {
zk.close();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Test
public void lock(){
//创建十个线程
for (int i = 0; i < 10; i++) {
new Thread(){
@Override
public void run() {
WatchCallBack watchCallBack = new WatchCallBack();
watchCallBack.setZk(zk);
String threadName = Thread.currentThread().getName();
watchCallBack.setThreadName(threadName);
//线程进行抢锁操作
watchCallBack.tryLock();
try {
//进行业务逻辑处理
System.out.println(threadName+" 开始处理业务逻辑了...");
Thread.sleep(200);
}catch (Exception e){
e.printStackTrace();
}
//释放锁
watchCallBack.unLock();
}
}.start();
}
while(true){
}
}
}
运行结果:
Thread-1 线程创建了一个节点为 : /lock0000000112
Thread-5 线程创建了一个节点为 : /lock0000000113
Thread-2 线程创建了一个节点为 : /lock0000000114
Thread-6 线程创建了一个节点为 : /lock0000000115
评论