15

zookeeper分布式锁（代码实现）

 3 years ago
source link: https://segmentfault.com/a/1190000037685014
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
neoserver,ios ssh client

前言

这里是zookeeper响应式编程的第二篇——自定义分布式锁，第一篇zookeeper分布式注册配置中心见如下链接：

https://segmentfault.com/a/11...

分布式锁

由于在分布式系统中，任意2个节点的线程需要获取同一个资源的时候就需要锁来维持程序的正确运行，但是呢，如果使用JVM提供的锁只能锁住自己，因为这是2台主机，这就引入了分布式锁的概念。也即是说需要一把锁在主机的外面，而很多框架都可以实现分布式锁，比如redis,mysql和zookeeper,目前最为方便的是使用zookeeper，因为就其高可用性和统一视图而言就比其他的技术方便很多。

对于zookeeper做分布式锁的分析过程如下：首先对于2台主机抢占同一把锁的时候，只能有一台主机成功抢占，那么有可能出现获得锁的主机“挂了”，那么我们可以使用临时节点解决该问题，那么在一个主机成功抢占该锁之后，如果它释放了锁，其他主机是如何知道它已经成功释放了呢？第一种方式就是可以采用主动轮询的方式不断检测该锁是否已经被释放，但是这种方式有延迟，并且在主机特别多的时候多台主机轮询一把锁会造成zookeeper很大的压力。第二种方式就是使用watch机制，这种方式可以解决延迟的问题，但是在获得锁的主机释放锁的时候，zookeeper会回调哪些所有没有抢到锁的线程，然后那些主机又会发起强锁的操作，会造成较大的通信压力。第三种方式就可以使用watche机制+序列节点，然后让每一个临时序列节点都watch前一个节点，这样只有一个编号最小的才能获得锁，并且在释放锁后会只通知后面的一个主机。

代码实现

首选我们需要在编写配置中心的Utils工具类，并且创建TestLock类实现分布式锁。然后我们开辟10个线程模拟多台主机抢占锁的过程，基本流程就是抢占锁，然后执行业务代码(这里使用睡眠来代替)，最后再释放锁。

Utils代码如下：

package org.qzx.config;

import org.apache.zookeeper.ZooKeeper;

import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 1:02 下午
 * @Description: org.qzx.config
 * @version: 1.0
 */
public class Utils {
    // zookeeper对象
    private static ZooKeeper zooKeeper;
    // 连接地址
    private static String address = "10.211.55.5:2181,10.211.55.8:2181,10.211.55.9:2181,10.211.55.10:2181/test";
    private static DefaultWatcher defaultWatcher = new DefaultWatcher();
    // 锁
    private static CountDownLatch latch = new CountDownLatch(1);

    public static ZooKeeper getZooKeeper() throws Exception{
        zooKeeper = new ZooKeeper(address,3000,defaultWatcher);
        defaultWatcher.setLatch(latch);
        latch.await();
        return zooKeeper;
    }
}

TestLock大体框架如下：

package org.qzx.lock;

import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.TimeUnit;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 3:54 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class TestLock {
    private ZooKeeper zooKeeper;

    @Before
    public void conn() throws Exception {
        zooKeeper = Utils.getZooKeeper();
    }

    @After
    public void close() throws InterruptedException {
        zooKeeper.close();
    }

    @Test
    public void TestLock(){
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                try {
                    // 抢占锁

                    // 业务代码
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName()+"is woorking!!!");
                    // 释放锁
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }).start();
        }
    }
}

我们在这里提供另外一个工具类可以为每一个线程实现抢锁和释放锁的过程，同时由于抢占的锁实际上是zookeeper的临时序列节点，所以必定会使用wather和回调机制，这里就把这个工具类叫做MyWatcherAndCallBack,该类提供抢占锁、释放锁，节点变化回调方法。其大体框架如下：

package org.qzx.lock;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 4:03 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class MyWatcherAndCallBack implements Watcher {
    private ZooKeeper zooKeeper;

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }

    @Override
    public void process(WatchedEvent event) {

    }

    // 抢占锁
    public void tryLock(){
        
    }

    // 释放锁
    public void unlock(){

    }
}

TestLock的代码也可以进行稍微的修改

package org.qzx.lock;

import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.TimeUnit;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 3:54 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class TestLock {
    private ZooKeeper zooKeeper;

    @Before
    public void conn() throws Exception {
        zooKeeper = Utils.getZooKeeper();
    }

    @After
    public void close() throws InterruptedException {
        zooKeeper.close();
    }

    @Test
    public void TestLock(){
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                MyWatcherAndCallBack myWatcherAndCallBack = new MyWatcherAndCallBack();
                myWatcherAndCallBack.setZooKeeper(zooKeeper);
                try {
                    // 抢占锁
                    myWatcherAndCallBack.tryLock();
                    // 业务代码
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName()+"is woorking!!!");
                    // 释放锁
                    myWatcherAndCallBack.unlock();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}
这样框架就已经搭建完毕，接下来就是编写具体的抢占锁和释放的逻辑代码了。

首先对于抢占锁的过程一定是阻塞的，直到抢占成功的时候才会接着往下走，这里使用CountDownLatch实现。而每一个线程都会创建属于自己的临时序列节点作为自己的锁，不过只有编号最小的那个才会获得被对应的线程所占有，其他的线程在创建节点后都会阻塞。这里为了方便看到那些线程创建了哪些锁，将线程的名字作为数据写入到节点中。然后我们在创建节点的回调函数中输出当前线程的名字和节点的名字，目的是为了检验代码写到现在是否正确。

MyWatcherAndCallBack代码如下：

package org.qzx.lock;

import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.TimeUnit;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 3:54 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class TestLock {
    private ZooKeeper zooKeeper;

    @Before
    public void conn() throws Exception {
        zooKeeper = Utils.getZooKeeper();
    }

    @After
    public void close() throws InterruptedException {
        zooKeeper.close();
    }

    @Test
    public void TestLock(){
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                MyWatcherAndCallBack myWatcherAndCallBack = new MyWatcherAndCallBack();
                myWatcherAndCallBack.setThreadName(Thread.currentThread().getName());
                myWatcherAndCallBack.setZooKeeper(zooKeeper);
                try {
                    // 抢占锁
                    myWatcherAndCallBack.tryLock();
                    // 业务代码
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName()+"is woorking!!!");
                    // 释放锁
                    myWatcherAndCallBack.unlock();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        while (true){

        }
    }
}

为了防止主线程执行太快导致回调函数还没有执行完毕就结束，在TestLock方法最后加上死循环进行阻塞。

@Test
public void TestLock(){
    for (int i = 0; i < 10; i++) {
        new Thread(()->{
            MyWatcherAndCallBack myWatcherAndCallBack = new MyWatcherAndCallBack();
            myWatcherAndCallBack.setThreadName(Thread.currentThread().getName());
            myWatcherAndCallBack.setZooKeeper(zooKeeper);
            try {
                // 抢占锁
                myWatcherAndCallBack.tryLock();
                // 业务代码
                TimeUnit.SECONDS.sleep(1);
                System.out.println(Thread.currentThread().getName()+"is woorking!!!");
                // 释放锁
                myWatcherAndCallBack.unlock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
    while (true){

    }
}

启动测试，在测试前记得创建工作目录/test，结果如下：

iuABVvi.png!mobile

可以看到节点创建的是有序的，但是线程是无序的。

接下来在创建节点成功的回调函数中，我们就需要获取锁了，使用getChildren方法获得工作目录下的孩子节点，也就是创建的临时序列节点，该方法不需要使用watch机制，因为不需要监测父节点，同时对于其回调对象我们也是同样封装在MyWatcherAndCallBack中。最后由于创建节点的名字在后面会用到，使用pathName属性保存当前线程创建的节点名字。

MyWatcherAndCallBack修改后代码如下：

package org.qzx.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 4:03 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class MyWatcherAndCallBack implements Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback {
    private ZooKeeper zooKeeper;
    private CountDownLatch latch = new CountDownLatch(1);
    private String threadName;
    private String pathName;

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }

    // Watcher
    @Override
    public void process(WatchedEvent event) {

    }

    // 抢占锁
    public void tryLock(){
        try {
            // 创建一个临时序列节点作为锁
            zooKeeper.create("/lcok",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                    this,"abc");
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 释放锁
    public void unlock(){

    }

    // StringCallback
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name!=null){
            System.out.println(threadName+"------>"+name);
            pathName = name;//类似于/lcok0000000000
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }

    //Children2Callback
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        
    }
}

在getChildren的回调方法中，当前线程一定成功创建了节点，并且可以看到所有在它之前创建的节点。那么我们现在遍历输出所有的children中的所有节点，目的是为了看到当前线程所看到的所有节点是无序的，这样就为后面需要排序提供了必要性。修改的部分代码如下：

//Children2Callback
@Override
public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
    System.out.println(threadName+"能看到的节点如下:");
    for (String child : children) {
        System.out.println(child);
    }
}

输出结果为：

uM7RJz7.png!mobile

可以看到第一个创建节点的线程为Thread-6,并且看到的节点都是无序的。并且节点的名字少了个/。

接下来就是对于当前线程，得判断它创建的锁是不是第一个，所以我们先对children进行排序，然后再获取当前锁在children的位置，如果是第一个说明该线程可以获得锁，执行latch.countdown(),这样该线程就可以去执行相应的任务了。如果不是第一个，那么就得判断前一个锁是否已经释放，判断的方法为exists，如果前面一个节点不存在了，说明已经释放，对于exists方法有可能会出现还没有成功监控到前一个节点就出现释放锁的情况，也就是exists执行失败了，没能监控前一个节点，那么说明锁已经释放，当前线程所需要进行的操作不在watcher中执行而是在回调函数中中执行，所以在这里exists的回调函数是必须的。

MyWatcherAndCallBack修改后的代码如下：

package org.qzx.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 4:03 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class MyWatcherAndCallBack implements Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback, AsyncCallback.StatCallback {
    private ZooKeeper zooKeeper;
    private CountDownLatch latch = new CountDownLatch(1);
    private String threadName;
    private String pathName;

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }

    // Watcher
    @Override
    public void process(WatchedEvent event) {

    }

    // 抢占锁
    public void tryLock(){
        try {
            // 创建一个临时序列节点作为锁
            zooKeeper.create("/lcok",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                    this,"abc");
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 释放锁
    public void unlock(){

    }

    // StringCallback
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name!=null){
            System.out.println(threadName+"------>"+name);
            pathName = name;//类似于/lcok0000000000
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }

    //Children2Callback
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        Collections.sort(children);
        int index = children.indexOf(pathName.substring(1));
        if(index==0){
            // 当前线程创建的锁是第一个，可以获得
            latch.countDown();
        }else {
            // 不是第一个，得判断前一个锁是否已经释放
            zooKeeper.exists("/"+children.get(index-1),this,this,"azz");
        }
    }

    // StatCallback
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {

    }
}

接下来需要对前一把锁的释放事件做处理，首先是在节点删除后，会触发节点删除时间，在该线程中会做出响应，具体做法就是要么直接latch.coutdown()获得锁或者通过getChildren判断当前锁是否是第一个了，是第一个就获得锁。同时对于当前线程获得锁后释放锁进行处理，直接对其创建的节点进行删除即可。

MyWatcherAndCallBack修改后的代码如下：

package org.qzx.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 4:03 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class MyWatcherAndCallBack implements Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback, AsyncCallback.StatCallback {
    private ZooKeeper zooKeeper;
    private CountDownLatch latch = new CountDownLatch(1);
    private String threadName;
    private String pathName;

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }

    // Watcher
    @Override
    public void process(WatchedEvent event) {
        switch (event.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                // 前一把锁被删除，当前线程获得锁
//                zooKeeper.getChildren("/",false,this,"aaa");
                latch.countDown();
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
        }
    }

    // 抢占锁
    public void tryLock(){
        try {
            // 创建一个临时序列节点作为锁
            zooKeeper.create("/lcok",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                    this,"abc");
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 释放锁
    public void unlock() throws KeeperException, InterruptedException {
        zooKeeper.delete(pathName,-1);// -1代表忽略版本号
    }

    // StringCallback
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name!=null){
            System.out.println(threadName+"------>"+name);
            pathName = name;//类似于/lcok0000000000
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }

    //Children2Callback
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        Collections.sort(children);
        int index = children.indexOf(pathName.substring(1));
        if(index==0){
            // 当前线程创建的锁是第一个，可以获得
            latch.countDown();
        }else {
            // 不是第一个，得判断前一个锁是否已经释放
            zooKeeper.exists("/"+children.get(index-1),this,this,"azz");
        }
    }

    // StatCallback
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {

    }
}

我们现在运行程序可以看到每一个线程都可以获得锁并且顺序执行。

Mn6ZZzQ.png!mobile

对于上述代码，存在一个问题，当前去除主线程业务代码中睡眠的操作，就会出现只有一个线程可以成功获得锁并且执行响应的操作，其他线程会出现类似于死锁的现象，但是这里不是死锁。这里的原因是执行速度太快了，很快就把当前线程获得的锁删除了，那么后面的线程在执行完排序，监控前面的锁就会出现失败的情况，这里的一种解决方法就是在exists的回调函数中针对节点不存在，也就是stat==null的时候，重新调用getChildren方法判断当前是否是第一把锁，如果是就会执行。

MyWatcherAndCallBack修改后的代码如下：

package org.qzx.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 4:03 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class MyWatcherAndCallBack implements Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback, AsyncCallback.StatCallback {
    private ZooKeeper zooKeeper;
    private CountDownLatch latch = new CountDownLatch(1);
    private String threadName;
    private String pathName;

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }

    // Watcher
    @Override
    public void process(WatchedEvent event) {
        switch (event.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                // 前一把锁被删除，当前线程获得锁
//                zooKeeper.getChildren("/",false,this,"aaa");
                latch.countDown();
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
        }
    }

    // 抢占锁
    public void tryLock(){
        try {
            // 创建一个临时序列节点作为锁
            zooKeeper.create("/lcok",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                    this,"abc");
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 释放锁
    public void unlock() throws KeeperException, InterruptedException {
        zooKeeper.delete(pathName,-1);
    }

    // StringCallback
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name!=null){
            System.out.println(threadName+"------>"+name);
            pathName = name;//类似于/lcok0000000000
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }

    //Children2Callback
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        Collections.sort(children);
        int index = children.indexOf(pathName.substring(1));
        if(index==0){
            // 当前线程创建的锁是第一个，可以获得
            latch.countDown();
        }else {
            // 不是第一个，得判断前一个锁是否已经释放
            zooKeeper.exists("/"+children.get(index-1),this,this,"azz");
        }
    }

    // StatCallback
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        if(stat==null){
            // 监控失败,自动获取锁
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }
}

在主线程的睡眠操作去除掉后，程序运行的结果如下：

BFzIbmz.png!mobile

可以看到所有线程又从新正常运行了。

到此，zookeeper自定义分布式锁的小demo就编写完毕。对于zookeeper分布式锁的所有代码整理如下。

Utils类：

package org.qzx.lock;

import org.apache.zookeeper.ZooKeeper;
import org.qzx.config.DefaultWatcher;

import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 1:02 下午
 * @Description: org.qzx.config
 * @version: 1.0
 */
public class Utils {
    // zookeeper对象
    private static ZooKeeper zooKeeper;
    // 连接地址
    private static String address = "10.211.55.5:2181,10.211.55.8:2181,10.211.55.9:2181,10.211.55.10:2181/test";
    private static DefaultWatcher defaultWatcher = new DefaultWatcher();
    // 锁
    private static CountDownLatch latch = new CountDownLatch(1);

    public static ZooKeeper getZooKeeper() throws Exception{
        zooKeeper = new ZooKeeper(address,3000,defaultWatcher);
        defaultWatcher.setLatch(latch);
        latch.await();
        return zooKeeper;
    }
}

TestLock类：

package org.qzx.lock;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.TimeUnit;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 3:54 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class TestLock {
    private ZooKeeper zooKeeper;

    @Before
    public void conn() throws Exception {
        zooKeeper = Utils.getZooKeeper();
    }

    @After
    public void close() throws InterruptedException {
        zooKeeper.close();
    }

    @Test
    public void TestLock(){
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                MyWatcherAndCallBack myWatcherAndCallBack = new MyWatcherAndCallBack();
                myWatcherAndCallBack.setThreadName(Thread.currentThread().getName());
                myWatcherAndCallBack.setZooKeeper(zooKeeper);
                try {
                    // 抢占锁
                    myWatcherAndCallBack.tryLock();
                    // 业务代码
//                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName()+"is woorking!!!");
                    // 释放锁
                    myWatcherAndCallBack.unlock();
                } catch (InterruptedException | KeeperException e) {
                    e.printStackTrace();
                }
            }).start();
        }
        while (true){

        }
    }
}

MyWatcherAndCallBack类：

package org.qzx.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Auther: qzx
 * @Date: 2020/10/29 - 10 - 29 - 4:03 下午
 * @Description: org.qzx.lock
 * @version: 1.0
 */
public class MyWatcherAndCallBack implements Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback, AsyncCallback.StatCallback {
    private ZooKeeper zooKeeper;
    private CountDownLatch latch = new CountDownLatch(1);
    private String threadName;
    private String pathName;

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }

    // Watcher
    @Override
    public void process(WatchedEvent event) {
        switch (event.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                // 前一把锁被删除，当前线程获得锁
//                zooKeeper.getChildren("/",false,this,"aaa");
                latch.countDown();
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
        }
    }

    // 抢占锁
    public void tryLock(){
        try {
            // 创建一个临时序列节点作为锁
            zooKeeper.create("/lcok",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
                    this,"abc");
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 释放锁
    public void unlock() throws KeeperException, InterruptedException {
        zooKeeper.delete(pathName,-1);
    }

    // StringCallback
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name!=null){
            System.out.println(threadName+"------>"+name);
            pathName = name;//类似于/lcok0000000000
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }

    //Children2Callback
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        Collections.sort(children);
        int index = children.indexOf(pathName.substring(1));
        if(index==0){
            // 当前线程创建的锁是第一个，可以获得
            latch.countDown();
        }else {
            // 不是第一个，得判断前一个锁是否已经释放
            zooKeeper.exists("/"+children.get(index-1),this,this,"azz");
        }
    }

    // StatCallback
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        if(stat==null){
            // 监控失败,自动获取锁
            zooKeeper.getChildren("/",false,this,"aaa");
        }
    }
}

report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK