基于zookeeper的分布式锁

网上看了一些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();

    }

 

}