寫在最前面

前幾周寫了篇 利用Redis實現分散式鎖 ，今天簡單總結下ZooKeeper實現分散式鎖的過程。其實生產上我只用過Redis或者資料庫的方式，之前還真沒了解過ZooKeeper怎麼實現分散式鎖。這周簡單寫了個小Demo，更堅定了我繼續使用Redis的信心了。

ZooKeeper分散式鎖的實現原理

在分散式解決方案中，Zookeeper是一個分散式協調工具。當多個JVM客戶端，同時在ZooKeeper上建立相同的一個臨時節點，因為臨時節點路徑是保證唯一，只要誰能夠建立節點成功，誰就能夠獲取到鎖。沒有建立成功節點，就會進行等待，當釋放鎖的時候，採用事件通知給客戶端重新獲取鎖資源。如果請求超時直接返回給客戶端超時，重新請求即可。

程式碼實現

為了更好的展現效果，我這裡設定每個執行緒請求需要1s，請求超時時間為30s。

 首先我們先寫一個測試類，模擬多執行緒多客戶端請求的情況：

public class ZkLockTest implements Runnable {
    private ZkLock zkLock = new ZkDistributedLock();

    public void run() {
        try {
            if (zkLock.getLock((long)30000,null)) {
                System.out.println(
 
"執行緒:" + Thread.currentThread().getName() + ",搶購成功:" + System.currentTimeMillis());
            } else {
                System.out.println("執行緒:" + Thread.currentThread().getName() + ",搶購超時失敗請重試:" + System.currentTimeMillis());
            }
            Thread.sleep(1000);
        } catch (Exception e) {

        } 
 
finally {
            zkLock.unLock();
        }
    }

    public static void main(String[] args) {
        System.out.println("zk分散式鎖開始。。");
        for (int i = 0; i < 100; i++) {
            new Thread(new ZkLockTest()).start();
        }
    }

}

模擬100個執行緒，去同時爭奪鎖。當然上述寫法 100個執行緒不會同時啟動，如果需要的話可以用訊號量的形式控制。

其次，寫一個鎖的介面

public interface ZkLock {

    // 獲取鎖
    Boolean getLock(Long acquireTimeout,Long endTime);

    // 釋放鎖
    void unLock();
}

這裡我定義了兩個介面，分別對應獲取鎖和釋放鎖。

在獲取鎖中有兩個引數，含義分別為鎖超時時間和最終計算的超時時間，具體看下文程式碼就懂了。

public class ZkDistributedLock implements ZkLock {
    // 叢集連線地址
    private String CONNECTION = "127.0.0.1:2181";
    // zk客戶端連線
    private ZkClient zkClient = new ZkClient(CONNECTION);
    // path路徑
    private String lockPath = "/lock";
    private CountDownLatch countDownLatch;
    //請求設定的超時時間：acquireTimeout 毫秒。最終超時時間endTime
    public Boolean getLock(Long acquireTimeout,Long endTime) {
        Boolean lock = false;
        if (endTime == null) {
            //等待超時時間
            endTime = System.currentTimeMillis() + acquireTimeout;
        }
        if (tryLock()) {
            System.out.println("####獲取鎖成功######");
            lock = true;
        } else {
            if (waitLock(endTime)) {
                if (getLock(null,endTime)) {
                    lock = true;
                }
            }
        }
       return lock;
    }

    public void unLock() {
        if (zkClient != null) {
            System.out.println("#######釋放鎖#########");
            zkClient.close();
        }
    }
    
    private boolean tryLock() {
        try {
            zkClient.createEphemeral(lockPath);
            return true;
        } catch (Exception e) {
            
            return false;
        }

    }

    private Boolean waitLock(Long endTime) {
       // System.out.println("進入等待");
        // 使用zk臨時事件監聽
        IZkDataListener iZkDataListener = null;
        try {
            // 使用zk臨時事件監聽
            iZkDataListener = new IZkDataListener() {
                
                public void handleDataDeleted(String path) throws Exception {
                    if (countDownLatch != null) {
                        countDownLatch.countDown();
                    }
                }
                public void handleDataChange(String arg0, Object arg1) throws Exception {
                    
                }
            };
            // 註冊事件通知
            zkClient.subscribeDataChanges(lockPath, iZkDataListener);
            if (System.currentTimeMillis() < endTime) {
                if (zkClient.exists(lockPath)) {
                    countDownLatch = new CountDownLatch(1);
                    try {
                        countDownLatch.await();
                        return true;
                    } catch (Exception e) {
                        
                    }
                } else {
                    return true;
                }
            } else {
                System.out.println("超時返回");
            }
        } catch (Exception e) {
            
        } finally {
            // 監聽完畢後，移除事件通知
            zkClient.unsubscribeDataChanges(lockPath, iZkDataListener);
        }
        return false;
    }

}

這個類是我實現zk鎖的核心類，和上文原理圖中類似。首先使用者請求的時候需要獲取鎖，第一個爭奪到鎖的使用者執行相關邏輯後釋放鎖，在這個過程中如果程式出錯斷開連線，因為臨時節點的緣故，節點也會自動刪除釋放鎖的。

另外就是其他爭奪鎖失敗的使用者，我這裡設定了一定的等待時間，當在時間內原鎖釋放，還是可以重新去獲取鎖的。這裡要說下鎖釋放的監聽，在原生的zookeeper中，使用watcher需要每次先註冊，而且使用一次就需要註冊一次。而在zkClient中，沒有註冊watcher的必要，而是引入了listener的概念，即只要client在某一個節點中註冊了listener，只要服務端發生變化，就會通知當前註冊listener的客戶端。我這裡使用的是IZkDataListener，這個類是zkClient提供的一個介面，它可以在當前節點資料內容或版本發生變化或者當前節點被刪除時觸發。

觸發後我們就可以重新去爭奪鎖，當再次爭奪失敗進入等待時會再次檢測當前請求是否超時。

下面我們來看下上述程式碼的實現效果：

zk分散式鎖開始。。
####獲取鎖成功######
執行緒:Thread-3,搶購成功:1544183770509
#######釋放鎖#########
####獲取鎖成功######
執行緒:Thread-81,搶購成功:1544183771555
#######釋放鎖#########

.........

超時返回
執行緒:Thread-11,搶購超時失敗請重試:1544183800677
超時返回
執行緒:Thread-1,搶購超時失敗請重試:1544183800681
#######釋放鎖#########
#######釋放鎖#########
####獲取鎖成功######
執行緒:Thread-49,搶購成功:1544183801710
超時返回
執行緒:Thread-25,搶購超時失敗請重試:1544183801729
超時返回
#######釋放鎖#########
#######釋放鎖#########

 釋放鎖說的可能並不準確，應該說是關閉連線，有些執行緒實際上是沒有得到鎖的。

簡單嘗試了下zk實現分散式鎖的方式，當然上述程式碼如果應用到生產中肯定問題還是不少的，因為興趣點不在這，就不仔細研究了。簡單來說，相比其他方式實現步驟更為複雜，感覺更容易出問題。

總結

經過三種方式的應用和簡單實踐，總結實現分散式鎖三種方式的優缺點如下

1、資料庫實現：

優點，實現簡單只是for update的顯示加鎖。缺點，效能問題較大，而且本身系統在設計時是需要儘量減輕資料庫的壓力的。

2、Redis實現：

優點：一般網際網路專案都會整合，本身是nosql資料庫，快取實現簡單，高併發應付自如，同時新版的Jedis完美解決了以往程式出錯，未設定超時時間死鎖的問題。

缺點：網路問題可能會引起鎖刪除失敗，超時時間有一定的延遲。

3、ZooKeeper實現：

優點：Zookeeper臨時節點先天可控的有效期設定，避免了程式引發的死鎖問題

缺點：實現過於繁雜，相比其他兩種寫法更容易出問題，另外還需要單獨維護zk。

結論：

我個人更為推薦Redis的實現方式，實現簡單，效能也比較好，同時引入叢集可以提高可用性。Jedis多參的設定方式也較好的保證了有效期的控制和死鎖的問題。