// 獲取互斥鎖
public void acquire() throws Exception;
// 在給定的時間內(nèi)獲取互斥鎖
public boolean acquire(long time, TimeUnit unit) throws Exception;
// 釋放鎖處理
public void release() throws Exception;
// 如果當(dāng)前線程獲取了互斥鎖，則返回true
boolean isAcquiredInThisProcess();

<dependency>
  <groupId>org.apache.logging.log4j</groupId>
  <artifactId>log4j-core</artifactId>
  <version>2.8.2</version>
</dependency>
<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.5.7</version>
</dependency>
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-framework</artifactId>
  <version>4.3.0</version>
</dependency>
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-recipes</artifactId>
  <version>4.3.0</version>
</dependency>
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-client</artifactId>
  <version>4.3.0</version>
</dependency>

package com.atguigu.case3;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorLockTest {

    public static void main(String[] args) {

        // 創(chuàng)建分布式鎖1
        InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");

        // 創(chuàng)建分布式鎖2
        InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock1.acquire();
                    System.out.println("線程1 獲取到鎖");

                    lock1.acquire();
                    System.out.println("線程1 再次獲取到鎖");

                    Thread.sleep(5 * 1000);

                    lock1.release();
                    System.out.println("線程1 釋放鎖");

                    lock1.release();
                    System.out.println("線程1  再次釋放鎖");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.acquire();
                    System.out.println("線程2 獲取到鎖");

                    lock2.acquire();
                    System.out.println("線程2 再次獲取到鎖");

                    Thread.sleep(5 * 1000);

                    lock2.release();
                    System.out.println("線程2 釋放鎖");

                    lock2.release();
                    System.out.println("線程2  再次釋放鎖");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    private static CuratorFramework getCuratorFramework() {

        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);

        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("xxx:2181,xxx:2181,xxx:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy).build();

        // 啟動客戶端
        client.start();

        System.out.println("zookeeper 啟動成功");
        return client;
    }
}

package com.rock.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * zk 分布式鎖 v1版本:
 * 完成功能 :
 *      1. 避免了驚群效應(yīng)
 * 缺失功能:
 *      1. 超時控制
 *      2. 讀寫鎖
 *      3. 重入控制
 */
public class DistributedLock {

    private String connectString;
    private int sessionTimeout;
    private ZooKeeper zk;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String waitPath;
    private String currentNode;
    private String LOCK_ROOT_PATH;

    private static String NODE_PREFIX = "w";

    public DistributedLock(String connectString, int sessionTimeout, String lockName){
        //TODO:數(shù)據(jù)校驗
        this.connectString = connectString;
        this.sessionTimeout = sessionTimeout;
        this.LOCK_ROOT_PATH = lockName;
    }


    public void init() throws IOException, KeeperException, InterruptedException {
        // 建聯(lián)
        zk = new ZooKeeper(connectString, sessionTimeout, watchedEvent -> {
            // connectLatch  連接上zk后  釋放
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                connectLatch.countDown();
            }
        });

        connectLatch.await();// 等待zk正常連接后

        // 判斷鎖名稱節(jié)點是否存在
        Stat stat = zk.exists(LOCK_ROOT_PATH, false);
        if (stat == null) {
            // 創(chuàng)建一下鎖名稱節(jié)點
            try {
                zk.create(LOCK_ROOT_PATH, LOCK_ROOT_PATH.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } catch (KeeperException e) {
                //并發(fā)創(chuàng)建沖突忽略。
                if (!e.code().name().equals("NODEEXISTS")) {
                    throw e;
                }
            }
        }
    }

    /**
     * 待補(bǔ)充功能:
     * 1. 超時設(shè)置
     * 2. 讀寫區(qū)分
     * 3. 重入控制
     */
    public void zklock() throws KeeperException, InterruptedException {
        if (!tryLock()) {
            waitLock();
            zklock();
        }
    }

    /**
     *
     */
    private void waitLock() throws KeeperException, InterruptedException {
        try {
            zk.getData(waitPath, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent){
                    // waitLatch  需要釋放
                    if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                        waitLatch.countDown();
                    }
                }
            }, new Stat());
            // 等待監(jiān)聽
            waitLatch.await();
        } catch (KeeperException.NoNodeException e) {
            //如果等待的節(jié)點已經(jīng)被清除了,不等了,再嘗試去搶鎖
            return;
        }

    }

    private boolean tryLock() throws KeeperException, InterruptedException {

        currentNode = zk.create(LOCK_ROOT_PATH + "/" + NODE_PREFIX, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 判斷創(chuàng)建的節(jié)點是否是最小的序號節(jié)點，如果是獲取到鎖；如果不是，監(jiān)聽他序號前一個節(jié)點
        List<String> children = zk.getChildren(LOCK_ROOT_PATH, false);
        // 如果children 只有一個值，那就直接獲取鎖； 如果有多個節(jié)點，需要判斷，誰最小
        if (children.size() == 1) {
            return true;
        } else {
            String thisNode = currentNode.substring(LOCK_ROOT_PATH.length() + 1);
            // 通過w00000000獲取該節(jié)點在children集合的位置
            int index = children.indexOf(thisNode);
            if (index == 0) {
                //自己就是第一個節(jié)點
                return true;
            }
            // 需要監(jiān)聽  他前一個節(jié)點變化
            waitPath = LOCK_ROOT_PATH + "/" + children.get(index - 1);
        }
        return false;
    }


    // 解鎖
    public void unZkLock(){
        // 刪除節(jié)點
        try {
            zk.delete(this.currentNode, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

}