基于zookeeper的分布式锁

a2005147

　　网上看了一些zookeeper对于分布式的实现，基本思路都是zookeeper顺序的临时节点，然后取最小的节点编号的方式，释放锁就是把这个临时节点删除，这个实现思路基本上正确的，但是有一些细节却可能出问题，zookeeper都是基于watcher的方式实现回调，watcher监听路径确很关键，有一种思路是watcher监听他的离他最近的节点路径，有一种思路是监听他的父亲节点路径，第一种方式在高并发的时候会非常大的问题，第一种方式的基本实现思路的伪代码如下
　　getchildren(父亲节点路径)
　　sort(list)
　　获取当前节点在list中的位置
　　if (位置0)
　　执行业务方法
　　否则
　　订阅  当前位置-1 的兄弟节点路径
　　这种方式最大的问题就是getchildren(父亲节点路径)和订阅之间存在时间差距，如果在高并发的时候有可能订阅的节点已经不存在，假设当前的位置为1，当他订阅位置0的时候，位置0的节点已经被其他进程删除了，如果这时候还订阅就等于永远都不会执行到回调函数里面，即使代码写在一起也有可能有问题，所以虽然第一种方式在消息的广播范围更小，而且对于zookeeper的压力更小，但是隐患还是很大
　　所以我采用第二种方式，代码如下，代码基于LockSupport可以实现
　　获得分布式锁
　　执行业务方法
　　释放分布式锁
　　这个流程比较符合编程习惯
　　增加一个状态控制，private volatile int state，控制多线程之间状态同步

　　通过先订阅后创建，
　　import java.util.Collections;
　　import java.util.List;
　　import java.util.concurrent.locks.LockSupport;
　　import org.I0Itec.zkclient.IZkChildListener;
　　import org.I0Itec.zkclient.ZkClient;
　　import org.apache.zookeeper.CreateMode;
　　public class DistributedClient {
　　private  String hosts = "zookeeper地址";
　　private String subNode = "sub";
　　private static String parentNode = "locks";
　　private   ZkClient zkclient = new ZkClient(hosts);
　　private String thisPath;
　　private Thread currentThread = null;
　　private IZkChildListener listener = null;
　　private volatile int state;
　　public void acqureDistributeLocks(final String groupNode) throws Exception {
　　final String parentNodePath = "/" + parentNode +"/"+groupNode;
　　if (!zkclient.exists(parentNodePath))
　　zkclient.createPersistent(parentNodePath);
　　currentThread = Thread.currentThread();
　　listener = new IZkChildListener() {
　　@Override
　　public void handleChildChange(String parentPath, List<String> currentChilds)
　　throws Exception {
　　if (thisPath == null)
　　          return;
　　List<String> childrenNodes = zkclient.getChildren(parentNodePath);
　　String thisNode = thisPath.substring((parentNodePath + "/").length());
　　Collections.sort(childrenNodes);
　　int index = childrenNodes.indexOf(thisNode);
　　if (index == 0) {
　　setState(1);
　　LockSupport.unpark(currentThread);
　　}
　　}
　　};
　　zkclient.subscribeChildChanges(parentNodePath, listener);
　　thisPath = zkclient.create(parentNodePath + "/" + subNode, null, CreateMode.EPHEMERAL_SEQUENTIAL);
　　if(getState() == 0) {
　　LockSupport.park(this);
　　}
　　}
　　public int getState() {
　　return state;
　　}
　　public void setState(int state) {
　　this.state = state;
　　}
　　public void releaseDistributeLocks(String groupNode) throws Exception {
　　String parentNodePath = "/" + parentNode +"/"+groupNode;
　　zkclient.unsubscribeChildChanges(parentNodePath, listener);
　　zkclient.delete(this.thisPath);
　　}
　　public static void main(String[] args) throws Exception {
　　for (int i = 0; i < 10; i++) {
　　new Thread() {
　　public void run() {
　　try {
　　DistributedClient dl = new DistributedClient();
　　dl.acqureDistributeLocks("test");
　　System.out.println(Thread.currentThread().getName() + " work finish");
　　dl.releaseDistributeLocks("test");
　　} catch (Exception e) {
　　e.printStackTrace();
　　}
　　}
　　}.start();
　　}
　　long begin = System.currentTimeMillis();
　　DistributedClient dl = new DistributedClient();
　　for (int i = 0; i < 100; i++) {
　　dl.acqureDistributeLocks("cnf");
　　System.out.println(Thread.currentThread().getName() + " work finish " + i);
　　dl.releaseDistributeLocks("cnf");
　　}
　　long end = System.currentTimeMillis();
　　System.out.println(end - begin);
　　System.in.read();
　　}
　　}