使用zookeeper以及curator客户端实现简单的分布式锁

1.分布式锁的实现

在分布式环境中，为了保证数据的一致性，经常在程序的某个运行点需要进行同步控制，接下来我们使用zookeeper来实现分布式锁：

分布锁的简单实现流程图如下：

zookeeper分布式锁流程图

1.多个客户端同时向zookeeper服务器的/lock节点下创建一个名字相同的临时顺序节点（顺序节点：是为了之后我们可以安装节点创建的的顺序来控制所得获取，越先创建的节点越早获取锁类似FIFO；临时节点：为什么是临时节点呢，考虑到如果当前的客户端由于某些原因导致挂掉了，那么他获取到的锁将不会被释放，那么导致的后果就是后续的节点无法再拿到锁，而临时节点则可以在客户端挂了后自动的删除该节点达到释放锁的效果）

2.客户端获取zookeeper服务/lock节点下的所有子节点，即所有等待获取锁的客户端

3.将上一步获取到的子节点按顺序进行排序（由于是顺序节点，因此他们的名字后一节是一次个递增的数字，我们可以按照数字将节点进行排序）

4.客户端通过比较判断是否当前的节点是数字最小的节点，如果是那么我们就认为有条件获取到锁如果没有我们需要进行第五步

5.由于我们不是数字最小的节点那么我们需要在比我们小1的节点上设置一个监听事件，监听节点的删除（达到的效果就是只要我的前一个节点是最小的那么他释放锁后我就会监听到，）

2.实现过程 

首先我们先定义一个zookeeper的操作类大致有的功能是如下：

2.1 按照上面的流程图我们首先需要在/lock下面创建节点，我们将其提取出来作为一个方法：

[code]/**
* 获取锁
*
* @return
* @throws Exception
*/
public void getLock() throws Exception {
String path = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
.forPath(BASELOCK + LOCK, "init".getBytes("UTF-8"));
this.tempOuPath = path;
}

在创建节点时需要通过withMode来只当节点类型，其类型有如下几种

• PERSISTENT(永久节点)

• PERSISTENT_SEQUENTIAL（永久顺序节点）

• EPHEMERAL（临时节点）

• EPHEMERAL_SEQUENTIAL（临时顺序节点）

2.2 通过第一步创建节点后我们需要获取/lock节点的子节点，进行排序比较判断是否当前节点可以拿到锁：

[code]public String toWaitLock() throws Exception {
Boolean isMin = false;

List<String> childs = client.getChildren().forPath( BASELOCK);
if (childs == null || childs.size() == 0) {
throw new RuntimeException("暂无可用的锁节点");
}

/**
* 将/lock下面的子节点按顺序排序
*/
childs = sortChild(childs);

/**
* 判断当前的节点是否时最小的节点
*/
isMin = isMinNode(childs);

if (!isMin) {
/**
* 如果不是最小节点，需要为前一个节点设置删除的监听事件，这里我们选择treeCache他可以监听所有的节点（父节点，子节点）的变化
*/
String finalOuPath = tempOuPath.substring(6,tempOuPath.length());
String preNode = childs.get(childs.indexOf(finalOuPath) - 1);
System.out.println("-------preNode---" + BASELOCK + "/" +preNode);
final TreeCache cache = new TreeCache(client,BASELOCK + "/" + childs.get(childs.indexOf(finalOuPath) - 1));
cache.getListenable().addListener(new TreeCacheListener() {
@Override
public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
System.out.println("now the eventType is : " + treeCacheEvent);
if(TreeCacheEvent.Type.NODE_REMOVED.equals(treeCacheEvent.getType())){
latch.countDown();
}
}
});
cache.start();
Listenable<TreeCacheListener> listenerListenable = cache.getListenable();
System.out.println(listenerListenable);
}else {
latch.countDown();
}

System.out.println("current path is getLock : " + isMin);
latch.await();

if (tempOuPath != null && !StringUtils.isEmpty(tempOuPath)) {
ouPath = tempOuPath;
}
return ouPath;
}

节点排序判断当前节点最小的方法：

[code]public List<String> sortChild(List<String> childs) {
Collections.sort(childs, new Comparator<String>() {
@Override
public int compare(String o1, String o2) {
String[] o1Arr = o1.split("-");
String[] o2Arr = o2.split("-");
return Integer.parseInt(o1Arr[1]) - (Integer.parseInt(o2Arr[1]));
}
});
return childs;
}

public boolean isMinNode(List<String> childs) {
String finalOuPath = tempOuPath.substring(6,tempOuPath.length());
if (childs.indexOf(finalOuPath) == 0) {
return true;
} else if (childs.indexOf(finalOuPath) < 0) {
throw new RuntimeException("there is no node for the path : " + tempOuPath);
} else {
return false;
}
}

释放锁，即删除客户端创建的节点

[code]public void releaseLock() throws Exception {
client.delete().forPath(ouPath);
}

编写客户端代码（客户端基本一致）：

[code]package com.lsm.clients.dispersedlock.client;

import com.lsm.clients.dispersedlock.ZookeeperUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;

import java.util.concurrent.CountDownLatch;

/**
* Created by lsm on 2018/7/22.
*/
public class Client1 {
private static CuratorFramework client1;
private static String zkServerPath = "192.168.139.145:2181";
private static CountDownLatch countDownLatch = new CountDownLatch(1);

/**
* 初始化客户端，并在初始化完成后让当前的线程等待
*
* @throws Exception
*/
public static void init() throws Exception {
RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
client1 = CuratorFrameworkFactory.builder()
.connectString(zkServerPath)
.sessionTimeoutMs(10000).retryPolicy(retryPolicy)
.namespace("workspace").build();
client1.start();
countDownLatch.await();
}

public static void main(String[] args) throws Exception {
/**
* 线程等待，以确保当前客户端不会断开连接
*/
new Thread(new Runnable() {
@Override
public void run() {
try {
init();
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
Thread.sleep(2000);
new Thread(new Runnable() {
@Override
public void run() {

ZookeeperUtils utils = new ZookeeperUtils(client1);
try {
utils.getLock();

String path = utils.toWaitLock();
if (path != null) {
System.out.println("the client1 get lock , path is" + path);

/**
* 获取锁后执行业务逻辑
*/
ZookeeperUtils.doSomething();

/**
* 释放锁
*/
utils.releaseLock();

/**
* 让线程恢复执行
*/
countDownLatch.countDown();
} else {
System.out.println("the client1 get lock failed");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
}

我们以三个客户端来进行测试：

客户端1先运行：由于他是/lock下面的第一个节点那么他会直接获得锁做下面的事情

可以看到客户端2再客户端释放锁后触发的删除的watcher事件而获取锁：

客户端3再客户端2释放锁 后获取到锁：

我们可以看到所有的客户端都是按照有锁的才能执行下面的逻辑，就好比我们的synchronize同一时间只能一个线程执行，到这里一个简单的分布式锁基本上完成了