基于Zookeeper使用ZkClient实现分布式锁

有段时间没写博客了，在整理之前写过的一套自定义框架，并且整理好上传值github上了，也有一些新功能还在开发，欢迎大家使用：一个好用的Http接口请求工具组件

可能今天这篇文章跟之前的比有些跳跃性，一下子就谈到了Zookeeper了，不过也没关系啦，先谈谈最常用，然后在慢慢看Zooeeper的其他知识。

简单介绍

ZooKeeper致力于提供一个高性能、高可用，且具备严格的顺序访问控制能力的分布式协调服务，是雅虎公司创建，是Google的Chubby一个开源的实现，也是Hadoop和Hbase的重要组件。它的数据以树形结构（类似于文件系统）储存在内存当中由于数储存在内存当中（并且每个节点都必须要有数据），所以拿取数据效率特别快。

分布式锁：功能与之前并发编程中的Lock功能一致，主要是为了解决共享资源被竞争所导致的并发问题。由于并发编程当中锁是在当前的JVM当中，而对于分布式的服务来说单纯的JVM的锁已经不起作用了，不过实现功能还是一致。

监听机制

既然是锁，那么就存在线程等待以及线程被唤醒功能，所以就需要有一个监听机制，当ZooKeeper上的锁被释放之后需监听到，并且通知服务去获取锁资源，正好在ZooKeeper当中存在一种监听机制，为事件监听器（Watcher）

事件监听器：客户端可以在节点上注册监听器，当特定的事件发生后，ZooKeeper会通知到感兴趣的客户端；被监听的事件有：NodeCreated（节点创建）、NodeDeleted（节点删除）、NodeDataChanged（节点数据被改变）、NodeChildrenChange（子节点被修改）

Node类型

基于ZooKeeper分布式锁必然基于节点，在ZooKeeper创建节点共有四点类型：

1、持久化节点（PERSISENT）：同一个节点路径只能创建一个节点，并且连接创建节点后，断开连接后，节点仍然存在并且关闭服务会保存至磁盘上。

2、持久化顺序节点（PERSISENT_SEQUENTIAL）：同一个节点路径可以创建多个节点，并且ZooKeeper会自动分配一个按顺序的节点号，断开连接后，节点仍然会保存至磁盘。

3、临时节点（EPHEMERAL）：在一个连接中，同一路径下只能创建一个节点，当创建节点的连接关闭后，该节点会被删除，如果非正常关闭连接，则过一段时间后节点会被删除。

4、临时顺序节点（EPHEMERAL_SEQUENTIAL）：同一路径下可以创建多个节点，但是节点名称ZooKeeper会自动分配一个按顺序的节点号，当连接关闭后，这些节点会被删除。

实现方式

既然是锁，那么必定是同一个节点，而且要先去尝试获取到锁，如果没有获取到，那么就进入等待，并且监听同一个节点是否被删除（或者修改），如果删除，则唤醒等待的线程，并且再次去获取ZooKeeper上的锁节点；那么整体的流程如下：

//锁节点
public static final String PATH = "/lock";
//尝试获取锁
public abstract boolean tryLock();
//等待锁释放
public abstract void waitLock();
//释放锁
public abstract void unLock();
//获取锁
void getLock(){if(tryLock()) {//获取到锁后，进行业务操作System.out.println(Thread.currentThread().getName() + " get Lock");}else {//没有获取则进入等待，并且监听锁节点是否被释放waitLock();//再次获取锁getLock();}
}

我这里是采用ZkClient进行操作ZooKeeper的，先创建一个ZkClient连接：

 CountDownLatch latch = null;private static final String CONNECTION = "127.0.0.1:2181";ZkClient zkClient = new ZkClient(CONNECTION,3000);

先来看看尝试获取锁可以怎样实现：

@Override
public boolean tryLock() {try {//创建临时节点，也可以创建持久化节点，到时候释放节点的时候删除就好了zkClient.createEphemeral(PATH, "1".getBytes());return true;} catch (Exception e) {//如果创建失败，则获取节点锁失败，则进入等待return false;}
}

进入等待，并且监听锁节点是否删除或者修改：

@Override
public void waitLock() {//创建监听事件IZkDataListener listen = new IZkDataListener() {public void handleDataChange(String dataPath, Object data) throws Exception {//当前方法为监听节点修改，如果节点进行修改，那么就会执行当前方法}public void handleDataDeleted(String dataPath) throws Exception {//我这里是释放锁为删除节点，删除会执行当前方法latch.countDown();}};//注册监听器zkClient.subscribeDataChanges(PATH, listen);//如果ZooKeeper上存在锁节点，那么进入等待if(zkClient.exists(PATH)) {//采用CountDownLatch等待latch = new CountDownLatch(1);try {//进入等待latch.await();} catch (InterruptedException e) {e.printStackTrace();}}//删除监听器zkClient.unsubscribeDataChanges(PATH, listen);
}

然后就是释放节点了：

@Override
public void unLock() {if(zkClient != null) {System.out.println(Thread.currentThread().getName() + " unlock.. ");//删除节点zkClient.delete(PATH);//关闭当前连接zkClient.close();}
}

这么一套流程下来，分布式锁的功能就完成了，当这种实现的功能类似JVM锁中的非公平锁，即没有先后顺序所言，如果想要达到公平锁，那么就必须得使用顺序节点进行操作了。

那么分布式公平锁监听的节点就不是同一个节点了，而是监听当前节点的上一个节点：

private String lockSeq = null;private String before = null;@Override
public boolean tryLock() {if(!zkClient.exists(PATH)) {zkClient.createPersistent(PATH, true);}if(lockSeq == null) {lockSeq = zkClient.createEphemeralSequential(PATH + "/", "1".getBytes());}List<String> children = zkClient.getChildren(PATH);Collections.sort(children);if(lockSeq.equals(PATH + "/" + children.get(0))) {return true;}else {for(String str : children) {if(lockSeq.contains(str)) {break;}before = PATH + "/" + str;}System.out.println(Thread.currentThread().getName() + " before node:" + before);return false;}
}@Override
public void waitLock() {IZkDataListener listen = new IZkDataListener() {public void handleDataChange(String dataPath, Object data) throws Exception {}public void handleDataDeleted(String dataPath) throws Exception {latch.countDown();}};zkClient.subscribeDataChanges(before, listen);if(zkClient.exists(before)) {latch = new CountDownLatch(1);try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}}zkClient.unsubscribeDataChanges(before, listen);}@Override
public void unLock() {if(zkClient != null) {System.out.println(Thread.currentThread().getName() + " unlock.. ");System.out.println("this node:" + lockSeq + " last node:" + before);zkClient.delete(lockSeq);zkClient.close();}
}

实现的代码如上，大家可以各位去试一下，个人觉得非公平锁的效率相比公平锁来说效率要高一点点，不过对应大量的分布式服务去竞争锁资源的话，个人建议还是使用公平锁，避免阻塞时间过长。