zookeeper 共享锁代码的实现

wxin

zookeeper存储是树状结构，如/path1/path2(这里path1必须是PERSISTENT，否则path2不能创建)，
节点有三种状态：
1、EPHEMERAL临时节点，server删除后该节点也会删除，但是我测试的时候，如果将server强制删除，EPHEMERAL节点会延迟删除；
2、PERSISTENT节点，这个是持久保存的节点。
3、SEQUENTIAL节点，自动标号的节点
如果父节点有子节点，父节点将不能删除。


watch是zookeeper最核心的一个功能；
watch = true;
监听某path1节点目录下是否创建删除子节点
zookeeper.getChildren("/path1", watch );
监听path2是否删除、修改
zookeeper.exists(path2,true);
zookeeper.getData(path2,true, null);




zookeeper能很好的协调分布式应用的任务执行，比如当多个应用同时处理分布式存储内存里面KEY VALUE、MQ时。锁只是其中一种表现。还有zookeeper 选举，集群管理，队列管理等；



public abstract class ZkLock implements Watcher {
private ZooKeeper zk;
private String hosts = "192.168.1.202:2181";
private String lockName = null;
private String nodePath = null;
private String nodeName = null;
private boolean extra = false;
private Integer SESSION_TIMEOUT = 5000;
private String waitPath = null;
public CountDownLatch latch = new CountDownLatch(1);

public ZkLock(String lockName,ZooKeeper zk) throws Exception{
this.lockName = StringUtils.contains(lockName, "/")?lockName:"/"+lockName;
if(zk==null){
  zk = new ZooKeeper(hosts, SESSION_TIMEOUT,this);
}
this.zk = zk;
if(zk.exists(this.lockName, false)==null){
zk.create(this.lockName, this.lockName.getBytes()
, Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
nodePath = zk.create(this.lockName + "/lock-", new byte[0],Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
nodeName = StringUtils.substringAfter(nodePath, this.lockName+"/");
}

public ZkLock(String lockName) throws Exception{
this(lockName,null);
}
@Override
public void process(WatchedEvent event) {
latch.countDown();
if (extra&&event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {
action();
        }
}

private Object exec() throws Exception{
List<String> list = zk.getChildren(lockName, false);
String[] nodes = list.toArray(new String[list.size()]);
Arrays.sort(nodes);
if(StringUtils.equals(nodeName, nodes[0])){
Object obj = action();
return obj;
}else{
return waitLock(nodes[0]);
}
}

private Object waitLock(String lower) throws Exception{
Stat stat = zk.exists(lockName + "/" + lower,true);
if(stat!=null){
latch.await();
}
return this.exec();
}

//这里的锁，主线程后面要加Thread.sleep()
private void extraExec() throws Exception{
        latch.await();
        List<String> childrenNodes = zk.getChildren(lockName, false);
        if (childrenNodes.size() == 1) {
        action();
        } else {
          Collections.sort(childrenNodes);
              int index = childrenNodes.indexOf(nodeName);
            if (index == -1) {
                // never happened
            } else if (index == 0) {
            action();
            } else {
//            System.out.println("waitPath:"+childrenNodes.get(index - 1));
            this.waitPath = this.lockName + "/" + childrenNodes.get(index - 1);
                // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
             if(zk.exists(waitPath, true)==null){
            action();
                 }
            }
        }
}

private Object action() {
try{
return doAction();
}finally{
try {
zk.delete(nodePath, -1);
//zk.close();
} catch (Exception e) {
throw new RuntimeException(e);
}

}
}

/**
* 主程必须守候,因为是与脱离主线程，所以无返回值。
* @return
* @throws Exception
*/
public void extraExecute() throws Exception{
this.extra = true;
this.extraExec();
}

public Object execute() throws Exception{
return exec();
}

public abstract Object doAction() ;


public static void main(String[] args) throws Exception {
Object obj = null;
final CountDownLatch startSignal = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {//zookeeper默认是一个应用10个连接
            new Thread() {
                public void run() {
                ZkLock lock;
try {
lock = new ZkLock("test") {
@Override
public Object doAction() {
return null;
}
};
//startSignal.await();
lock.execute();
                                              //  lock.extraExecute();
                                      //Thread.sleep(Long.MAX_VALUE);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
                }
            }.start();
            startSignal.countDown();
        }
}

}