ZooKeeper实现分布式锁

Lock

publicinterfaceLock{publicvoidlock();publicvoidunlock();}

ZkLock

/**
 * 使用ZooKeeper实现分布式锁
 * */publicclassZkLockimplementsLock{privateCuratorFramework client;privatefinalString zkPath;privateInteger count;// 记录获取锁的次数，实现可重入锁privatefinalString subNodePathPrefix;privateString lockedPath;privateString preNodePath;privateString subShortPath;privateThread thread;publicZkLock(CuratorFramework client,String zkPath)throwsException{this.client = client;this.zkPath = zkPath;
        subNodePathPrefix ="node-";
        count =0;init();}privatevoidinit()throwsException{synchronized(ZkLock.class){Stat stat;
            stat = client.checkExists().forPath(zkPath);if(stat ==null)
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(zkPath);}}@Overridepublicvoidlock(){if(thread ==Thread.currentThread()){// 实现可重入锁
            count +=1;return;}if(lockInternal()){
            thread =Thread.currentThread();
            count +=1;}}/**
     * 上锁函数
     * */privatebooleanlockInternal(){try{boolean locked =tryLock();// 尝试上锁if(locked)returntrue;// 上锁失败，阻塞等待while(!locked)
                locked =await();returntrue;}catch(Exception e){
            e.printStackTrace();}returnfalse;}privatebooleantryLock()throwsException{// 为此上锁请求建立临时有序节点
        lockedPath = client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(zkPath +"/"+ subNodePathPrefix);if(lockedPath ==null){thrownewException();}// 获取锁节点下面所有的上锁请求节点，判断该上锁请求节点的位置，若为首位则获得锁，否则尝试上锁失败
        subShortPath =getShortPath(lockedPath);List<String> waiters =getWaiters();if(checkIsHeadNode(waiters))returntrue;int index =Collections.binarySearch(waiters, subShortPath);if(index <0)thrownewException();
        preNodePath = zkPath +"/"+ waiters.get(index -1);returnfalse;}privateStringgetShortPath(String path){int index = path.lastIndexOf(zkPath +"/");if(index >=0){
            index += zkPath.length()+1;return index <= path.length()? path.substring(index):"";}returnnull;}privatebooleancheckIsHeadNode(List<String> waiters){Collections.sort(waiters);return subShortPath.equals(waiters.get(0));}privateList<String>getWaiters()throwsException{return client.getChildren().forPath(zkPath);}privatebooleanawait()throwsException{if(preNodePath ==null)thrownewException();CountDownLatch latch =newCountDownLatch(1);
        client.getData().usingWatcher((Watcher) watchedEvent -> latch.countDown()).forPath(preNodePath);
        latch.await();returntrue;}/**
     * 解锁函数
     * */@Overridepublicvoidunlock(){if(!thread.equals(Thread.currentThread()))return;// 检查重入次数
        count -=1;if(count >0)return;try{
            client.delete().forPath(lockedPath);}catch(Exception e){
            e.printStackTrace();}}}