一、前言

在同一個jvm程序中時，可以使用JUC提供的一些鎖來解決多個執行緒競爭同一個共享資源時候的執行緒安全問題，但是當多個不同機器上的不同jvm程序共同競爭同一個共享資源時候，juc包的鎖就無能無力了，這時候就需要分散式鎖了。常見的有使用zk的最小版本，redis的set函式，資料庫鎖來實現，本節我們談談使用zookeeper的序列節點機制來實現一個分散式鎖。

二、使用zookeeper實現分散式鎖

首先我們先來看看使用zk實現分散式鎖的原理，在zk中是使用檔案目錄的格式存放節點內容，其中節點型別分為：

• 持久節點（PERSISTENT ）：節點建立後，一直存在，直到主動刪除了該節點。
• 臨時節點（EPHEMERAL）：生命週期和客戶端會話繫結，一旦客戶端會話失效，這個節點就會自動刪除。
• 序列節點（SEQUENTIAL ）：多個執行緒建立同一個順序節點時候，每個執行緒會得到一個帶有編號的節點，節點編號是遞增不重複的，如下圖：

如上圖，三個執行緒分別建立路徑為/root/node的節點，可知在zk伺服器端會在root下存在三個node節點，並且器編號唯一遞增。

具體在節點建立過程中，可以混合使用，比如臨時順序節點（EPHEMERAL_SEQUENTIAL），這裡我們就使用臨時順序節點來實現分散式鎖。

分散式鎖實現：

• 建立臨時順序節點,比如/root/node，假設返回結果為nodeId。
• 獲取/root下所有孩子節點，用自己建立的nodeId的序號與所有子節點比較，看看自己是不是編號最小的。如果是最小的則就相當於獲取到了鎖，如果自己不是最小的，則從所有子節點裡面獲取比自己次小的一個節點，然後設定監聽該節點的事件，然後掛起當前執行緒。
• 當最小編號的執行緒獲取鎖，處理完業務後刪除自己對應的nodeId，刪除後會啟用比自己大一號的節點的執行緒從阻塞變為執行態，被啟用的執行緒應該就是當前node序列號最小的了，然後就會獲取到鎖。

下面我們看看程式碼實現：

public class ZookeeperDistributedLock {
    public final static Joiner j = Joiner.on("|").useForNull("");

    //zk客戶端
    private ZooKeeper zk;
    //zk是一個目錄結構，root為最外層目錄
    private String root = "/locks"
 
;
    //鎖的名稱
    private String lockName;
    //當前執行緒建立的序列node
    private ThreadLocal<String> nodeId = new ThreadLocal<>();
    //用來同步等待zkclient連結到了服務端
    private CountDownLatch connectedSignal = new CountDownLatch(1);
    private final static int sessionTimeout = 3000;
    private final static byte[] data= new byte[0];

    public ZookeeperDistributedLock(String config, String lockName) {
        this.lockName = lockName;

        try {
            zk = new ZooKeeper(config, sessionTimeout, new Watcher() {

                @Override
                public void process(WatchedEvent event) {
                    // 建立連線
                    if (event.getState() == KeeperState.SyncConnected) {
                        connectedSignal.countDown();
                    }
                }

            });
            
            connectedSignal.await();
            Stat stat = zk.exists(root, false);
            if (null == stat) {
                // 建立根節點
                zk.create(root, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    class LockWatcher implements Watcher {
        private CountDownLatch latch = null;

        public LockWatcher(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void process(WatchedEvent event) {

            if (event.getType() == Event.EventType.NodeDeleted)
                latch.countDown();
        }
    }

    public void lock() {
        try {

            // 建立臨時子節點
            String myNode = zk.create(root + "/" + lockName , data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            System.out.println(j.join(Thread.currentThread().getName() + myNode, "created"));

            // 取出所有子節點
            List<String> subNodes = zk.getChildren(root, false);
            TreeSet<String> sortedNodes = new TreeSet<>();
            for(String node :subNodes) {
                sortedNodes.add(root +"/" +node);
            }
            
            String smallNode = sortedNodes.first();
            String preNode = sortedNodes.lower(myNode);

            if (myNode.equals( smallNode)) {
                // 如果是最小的節點,則表示取得鎖
                System.out.println(j.join(Thread.currentThread().getName(), myNode, "get lock"));
                this.nodeId.set(myNode);
                return;
            }

            CountDownLatch latch = new CountDownLatch(1);
            Stat stat = zk.exists(preNode, new LockWatcher(latch));// 同時註冊監聽。
            // 判斷比自己小一個數的節點是否存在,如果不存在則無需等待鎖,同時註冊監聽
            if (stat != null) {
                System.out.println(j.join(Thread.currentThread().getName(), myNode,
                        " waiting for " + root + "/" + preNode + " released lock"));

                latch.await();// 等待，這裡應該一直等待其他執行緒釋放鎖
                nodeId.set(myNode);
                latch = null;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    public void unlock() {
        try {
            System.out.println(j.join(Thread.currentThread().getName(), nodeId.get(), "unlock "));
            if (null != nodeId) {
                zk.delete(nodeId.get(), -1);
            }
            nodeId.remove();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

}

ZookeeperDistributedLock的建構函式建立zkclient，並且註冊了監聽事件，然後呼叫connectedSignal.await()掛起當前執行緒。當zkclient連結到伺服器後，會給監聽器傳送SyncConnected事件，監聽器判斷當前連結已經建立了，則呼叫 connectedSignal.countDown();啟用當前執行緒，然後建立root節點。

獲取鎖的方法lock，內部首先建立/root/lockName的順序臨時節點，然後獲取/root下所有的孩子節點，並對子節點進行排序，然後判斷自己是不是最小的編號，如果是直接返回true標示獲取鎖成功。否者看比自己小一個號的節點是否存在，存在則註冊該節點的事件，然後掛起當前執行緒，等待比自己小一個數的節點釋放鎖後傳送節點刪除事件，事件裡面啟用當前執行緒。

釋放鎖的方法unlock比較簡單，就是簡單的刪除獲取鎖時候建立的節點。

三、總結

本文使用zk的臨時順序節點以及節點事件通知機制實現了一個分散式鎖，大家想想是否還有優化的空間，我知道的還有一個改進點，另外一個思考，為何要使用臨時節點那？期待下回分解，

最後

想了解JDK NIO和更多Netty基礎的可以單擊我

想了解更多關於粘包半包問題單擊我
更多關於分散式系統中服務降級策略的知識可以單擊 單擊我
想系統學dubbo的單擊我
想學併發的童鞋可以 單擊我

加多

高階 Java 攻城獅 at 阿里巴巴加多，目前就職於阿里巴巴，熱衷併發程式設計、ClassLoader,Spring等開源框架,分散式RPC框架dubbo，springcloud等；愛好音樂，運動。微信公眾號：技術原始積累。知識星球賬號：技術原始積累