Zookeeper Java 开发，自定义分布式锁示例

文章目录

- 一、概述
- 二、导入依赖包
- 三、创建锁的过程
- - 3.1 通过 create 创建节点信息
  - 3.2 AsyncCallback.StringCallback 回调函数
  - 3.3 AsyncCallback.Children2Callback 的回调函数
  - 3.4 Watcher 的回调函数
- 四、完整示例
- - 4.1 完整分布式锁代码
  - 4.2 测试类

如果您还没有安装Zookeeper请看ZooKeeper 安装说明，Zookeeper 命令使用方法和数据说明，Zookeeper Java 开发入门。

一、概述

情景：假设有10个客户端（分散的10台主机）要执行一个任务，这个任务某些过程需要保持原子性。那么我们就需要一个分布式锁。
原理：通过在Zookeeper中创建序列节点来实现获得锁，删除节点来释放锁。其实质是一个按先来后到的排序过程，实现过程如下：
- 客户端发起请求，创建锁序列节点（/lock/xxxxxxxx）
- 获取所有锁节点，判断自己是否为最小节点
  - 如果自己是最小序列节点，则立即获得锁。
  - 否则不能获得锁，但要监控前一个序列节点的状态。
- 获得锁的客户端开始执行任务。
- 执行完任务后释放锁。
  - 由于后一个节点监控了前一个节点，当前一个节点删除时，后一个客户端会收到回调。
  - 在这个回调中再获取所有锁节点，判断自己是否为最小节点。
  - 以此类推，直到全部结束。
流程如下

在这里插入图片描述

如果您对没有做过 Zookeeper 开发，强烈建立先看 Zookeeper Java 开发入门。

二、导入依赖包

在 pom.xml 文件中导入 Zookeeper 包，注意一般这个包的版本要和您安装的 Zookeeper 服务端版本一致。
```
<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.8.2</version>
</dependency>
```

三、创建锁的过程

3.1 通过 create 创建节点信息

通过 create 创建序列节点信息。他是异步方式，创建成功后会调用 AsyncCallback.StringCallback.processResult 回调函数。

    public void lock() throws InterruptedException, LockException {zooKeeper.create("/lock", "xxx".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, context);countDownLatch.await();if(StringUtils.isEmpty(this.lockNodePath)){throw new LockException("创建锁失败");}System.out.println(this.appName + " 获得锁");}

3.2 AsyncCallback.StringCallback 回调函数

在 AsyncCallback.StringCallback 的回调函数中通过 getChildren 方法获取 ZooKeeper 锁节点下的所有节点信息。这个方法是异步的，调用成功后会调用 AsyncCallback.Children2Callback.processResult 回调函数。

    // AsyncCallback.StringCallback@Overridepublic void processResult(int i, String s, Object o, String s1) {if(StringUtils.isEmpty(s1)){// 这里是创建锁失败的情况。this.countDownLatch.countDown();return;}System.out.println(this.appName + " create lock node="+s1);this.lockNodePath = s1;// 获取 ZooKeeper 锁节点下的所有节点信息，以此来判断我是不是第一个创建节点，如果就获得锁，否则监控前一个节点信息。zooKeeper.getChildren("/", false, this, context);}

3.3 AsyncCallback.Children2Callback 的回调函数

在 AsyncCallback.Children2Callback 的回调函数判断我是不是第一个创建节点，如果就获得锁，否则监控前一个节点信息。监控前一个节点信息使用 exists 方法，这个方法设置了 Watcher 的 processResult 回调函数

    // AsyncCallback.Children2Callback@Overridepublic void processResult(int i, String s, Object o, List<String> list, Stat stat) {Collections.sort(list);//        for (String s1 : list) {
//            System.out.println("\t "+this.lockNodePath+" previous lock node="+s1);
//        }int index = list.indexOf(lockNodePath.substring(1));if(0 == index){// 如果我现在是第一个节点，则获得锁try {zooKeeper.setData("/", this.lockNodePath.getBytes(), -1);} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}this.countDownLatch.countDown();}else {// 我不是第一个节点，监控前一个节点信息（等他删除后我就是可能是第一个了）String watchNodePath = "/" + list.get(index - 1);System.out.println("\t "+this.lockNodePath+" watch node："+ watchNodePath);zooKeeper.exists(watchNodePath, this, new StatCallback() {@Overridepublic void processResult(int i, String s, Object o, Stat stat) {}}, context);}}

3.4 Watcher 的回调函数

在 Watcher 的回调函数，我们通过判断 watchedEvent.getType() 为 NodeDeleted 类型时，重新获取 ZooKeeper 锁节点下的所有节点信息，这使得消息回到了 “3.3”步，判断谁是第一个节点，然后获得得，完成整个流程。

    // Watcher@Overridepublic void process(WatchedEvent watchedEvent) {switch (watchedEvent.getType()) {case None:break;case NodeCreated:break;case NodeDeleted:zooKeeper.getChildren("/", false, this, context);break;case NodeDataChanged:break;case NodeChildrenChanged:break;case DataWatchRemoved:break;case ChildWatchRemoved:break;case PersistentWatchRemoved:break;}}

四、完整示例

4.1 完整分布式锁代码

package top.yiqifu.study.p131;import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class ZookeeperLock implements Watcher, AsyncCallback.StringCallback,AsyncCallback.Children2Callback {private String appName;private ZooKeeper zooKeeper;private Object context;private String lockNodePath;private CountDownLatch countDownLatch = new CountDownLatch(1);public ZookeeperLock(String name, ZooKeeper zk){this.appName = name;this.zooKeeper = zk;this.context = this;}public void lock() throws InterruptedException, LockException {zooKeeper.create("/lock", "xxx".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, context);countDownLatch.await();if(StringUtils.isEmpty(this.lockNodePath)){throw new LockException("创建锁失败");}System.out.println(this.appName + " 获得锁");}public void unlock() throws KeeperException, InterruptedException, LockException {if(StringUtils.isEmpty(this.lockNodePath)){throw new LockException("没有获得锁，无法释放锁");}zooKeeper.delete(lockNodePath, -1);System.out.println(this.appName + " 释放锁");}// AsyncCallback.StringCallback@Overridepublic void processResult(int i, String s, Object o, String s1) {if(StringUtils.isEmpty(s1)){// 这里是创建锁失败的情况。this.countDownLatch.countDown();return;}System.out.println(this.appName + " create lock node="+s1);this.lockNodePath = s1;// 获取 ZooKeeper 锁节点下的所有节点信息，以此来判断我是不是第一个创建节点，如果就获得锁，否则监控前一个节点信息。zooKeeper.getChildren("/", false, this, context);}// AsyncCallback.Children2Callback@Overridepublic void processResult(int i, String s, Object o, List<String> list, Stat stat) {Collections.sort(list);//        for (String s1 : list) {
//            System.out.println("\t "+this.lockNodePath+" previous lock node="+s1);
//        }int index = list.indexOf(lockNodePath.substring(1));if(0 == index){// 如果我现在是第一个节点，则获得锁try {zooKeeper.setData("/", this.lockNodePath.getBytes(), -1);} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}this.countDownLatch.countDown();}else {// 我不是第一个节点，监控前一个节点信息（等他删除后我就是可能是第一个了）String watchNodePath = "/" + list.get(index - 1);System.out.println("\t "+this.lockNodePath+" watch node："+ watchNodePath);zooKeeper.exists(watchNodePath, this, new StatCallback() {@Overridepublic void processResult(int i, String s, Object o, Stat stat) {}}, context);}}// Watcher@Overridepublic void process(WatchedEvent watchedEvent) {switch (watchedEvent.getType()) {case None:break;case NodeCreated:break;case NodeDeleted:zooKeeper.getChildren("/", false, this, context);break;case NodeDataChanged:break;case NodeChildrenChanged:break;case DataWatchRemoved:break;case ChildWatchRemoved:break;case PersistentWatchRemoved:break;}}public class LockException extends  Exception{public LockException(String message){super(message);}}
}

4.2 测试类

package top.yiqifu.study.p131;import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class Test06_ZookeeperLock {public static void main(String[] args) {try {// 创建 ZooKeeper 对象final ZooKeeper zooKeeper = testCreateZookeeper();int clientCount = 10;final CountDownLatch countDownLatch = new CountDownLatch(clientCount);for (int i = 0; i < clientCount; i++) {new Thread(new Runnable(){@Overridepublic void run() {TestLock(zooKeeper);countDownLatch.countDown();}}).start();}countDownLatch.await();} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}private static void  TestLock(ZooKeeper zooKeeper){try {String appName = Thread.currentThread().getName();ZookeeperLock zookeeperLock = new ZookeeperLock(appName, zooKeeper);// 加锁（获得分布式锁）zookeeperLock.lock();System.out.println(appName + " 执行任务");Thread.sleep(1000);// 释放锁zookeeperLock.unlock();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();} catch (ZookeeperLock.LockException e) {e.printStackTrace();}}private static ZooKeeper testCreateZookeeper() throws IOException, InterruptedException, KeeperException {final CountDownLatch countDownLatch = new CountDownLatch(1);// ZooKeeper 集群地址（没连接池的概念，是Session的概念）//String connectionString = "192.168.8.51:2181,192.168.8.52:2181,192.168.8.53:2181";String connectionString = "192.168.8.51:2181,192.168.8.52:2181,192.168.8.53:2181/aaa";// ZooKeeper Session 数据超时时间（也就是这个Session关闭后，与这个Session相关的数据在ZooKeeper中保存多信）。Integer sessionTimeout = 3000;// ZooKeeper Session 级别 Watcher（Watch只发生在读方法上，如 get、exists）final ZooKeeper zooKeeper = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {try {Event.KeeperState state = watchedEvent.getState();Event.EventType type = watchedEvent.getType();String path = watchedEvent.getPath();switch (state) {case Unknown:break;case Disconnected:break;case NoSyncConnected:break;case SyncConnected:countDownLatch.countDown();break;case AuthFailed:break;case ConnectedReadOnly:break;case SaslAuthenticated:break;case Expired:break;case Closed:break;}switch (type) {case None:break;case NodeCreated:break;case NodeDeleted:break;case NodeDataChanged:break;case NodeChildrenChanged:break;case DataWatchRemoved:break;case ChildWatchRemoved:break;case PersistentWatchRemoved:break;}System.out.println("Session watch state=" + state);System.out.println("Session watch type=" + type);System.out.println("Session watch path=" + path);} catch (Exception e) {e.printStackTrace();}}});countDownLatch.await();ZooKeeper.States state = zooKeeper.getState();switch (state) {case CONNECTING:break;case ASSOCIATING:break;case CONNECTED:break;case CONNECTEDREADONLY:break;case CLOSED:break;case AUTH_FAILED:break;case NOT_CONNECTED:break;}System.out.println("ZooKeeper state=" + state);return zooKeeper;}}