zookeeper知识结构3-分布式锁

zkcli    连接默认zookeeper服务器

zkcli -server ip:port    连接指定的zookeeper服务器

create -s -e path data [acl]    创建节点，-s表示顺序，-e表示临时，默认是持久节点，acl缺省表示不做任何权限限制

ls path [watch]    显示path下的节点，不递归显示,watch注册监听，命令行可忽视

ls2 path    显示当前节点下的节点和当前节点的属性信息

get path [watch]    获取path的属性信息和数据内容

set path data [version]    更新path的数据内容，version是做类似cas的功能的对应dataversion，命令行可忽略

delete path [version]    删除节点，不能递归删除，只能删除叶子节点

setacl path acl    设置节点acl,例子(scheme:id:password=:perm)-(digest:example:sha-1(base64(pwd))=:cdrwa) create delete read write admin

getacl path    获取path节点的acl

stat path    查看path的属性信息

quit 退出zkcli

/**
 * 尝试加锁,直接创建节点，如果节点创建失败，说明加锁失败
 * @param lockName
 * @return
 */
public boolean tryLock(String lockName) {
    try {
        //创建节点
        String path = cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(getLockPath(lockName), lockName.getBytes());
       logger.info(Thread.currentThread().getName()+ "try lock  success ,path:"+path+" tid:"+Thread.currentThread().getName());
        return true;
    }catch (KeeperException.NodeExistsException e) {
       logger.info("try lock fail,"+" tid:"+Thread.currentThread().getName());
    }catch (Exception ex) {
        ex.printStackTrace();
    }
    return false;
}

/**
 * 尝试加锁失败后，阻塞等待
 * @param lockName
 * @throws Exception
 */
public void waitForLock(String lockName) throws Exception {
    //监听子节点
    PathChildrenCache pathChildrenCache = new PathChildrenCache(cf, LOCK_PATH, false);
    pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
    pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
            switch (event.getType()) {
                case CHILD_REMOVED:
                    logger.info("path:" + event.getData().getPath() + " has removed,start to lock:{}",Thread.currentThread().getName());
                    countDownLatch.countDown();
                    break;
                default:
                    //logger.info(" has changeed,{},start to lock:{}",event.getType(),Thread.currentThread().getName());
                    break;
            }
        }
    });
    boolean hasLock = false;
    while(!hasLock) {
        Stat stat = cf.checkExists().forPath(getLockPath(lockName));
        //节点存在,此处与unlock非原子操作，如果在checkExists返回true时刻，成功unlock,那此端无限等待
        if (stat == null) {
            logger.info("waitForLock not exists:{}", Thread.currentThread().getName());
           hasLock = tryLock(lockName);

           if(hasLock) {
               logger.info("waitForLock get lock :{}", Thread.currentThread().getName());
               break;
           }
        } else {
            logger.info("waitForLock exists:{}", Thread.currentThread().getName());
            countDownLatch.await();
        }
    }
}

public boolean unlock(String lockName) throws Exception {
        logger.info("start unlock:" + lockName + " tid:" + Thread.currentThread().getName());
        cf.delete().forPath(getLockPath(lockName));
        logger.info("end unlock:" + lockName + " tid:" + Thread.currentThread().getName());
    return true;
}

public void waitForLock(String lockName) throws Exception {
    logger.info("waitForLock {}:{}",beforeNode, Thread.currentThread().getName());
    boolean hasLock = false;
    while(!hasLock) {
       //是不是最小节点
        boolean isMin = isMinNode();
        if (isMin) {//是 则成功获取锁
            logger.info("waitForLock getLock:{}", Thread.currentThread().getName());
            break;
        } else {
            try {
                cf.getData().usingWatcher(new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        switch (event.getType()) {
                            case NodeDeleted:
                                logger.info("path:" + event.getPath() + " has removed,before:{},start to lock:{}", beforeNode, Thread.currentThread().getName());
                                countDownLatch.countDown();
                                break;
                        }
                    }
                }).forPath(beforeNode);
                logger.info("waitForLock waiting:{}", Thread.currentThread().getName());
                countDownLatch.await();
                logger.info("开始抢占:{},{}", Thread.currentThread().getName(), currentNode);
            }catch (KeeperException.NoNodeException e){
                logger.info("waitForLock has delete:{},{}",beforeNode, Thread.currentThread().getName());
            }
        }
    }
}