基于Zookeeper的分布式鎖

這篇文章隻需要你10分鐘的時間。

實作分布式鎖目前有三種流行方案，分别為基于資料庫、Redis、Zookeeper的方案，其中前兩種方案網絡上有很多資料可以參考，本文不做展開。我們來看下使用Zookeeper如何實作分布式鎖。

什麼是Zookeeper?

Zookeeper(業界簡稱zk)是一種提供配置管理、分布式協同以及命名的中心化服務，這些提供的功能都是分布式系統中非常底層且必不可少的基本功能，但是如果自己實作這些功能而且要達到高吞吐、低延遲同時還要保持一緻性和可用性，實際上非常困難。是以zookeeper提供了這些功能，開發者在zookeeper之上建構自己的各種分布式系統。

雖然zookeeper的實作比較複雜，但是它提供的模型抽象卻是非常簡單的。Zookeeper提供一個多層級的節點命名空間(節點稱為znode)，每個節點都用一個以斜杠(/)分隔的路徑表示，而且每個節點都有父節點(根節點除外)，非常類似于檔案系統。例如，/foo/doo這個表示一個znode，它的父節點為/foo，父父節點為/，而/為根節點沒有父節點。與檔案系統不同的是，這些節點都可以設定關聯的資料，而檔案系統中隻有檔案節點可以存放資料而目錄節點不行。Zookeeper為了保證高吞吐和低延遲，在記憶體中維護了這個樹狀的目錄結構，這種特性使得Zookeeper不能用于存放大量的資料，每個節點的存放資料上限為1M。

而為了保證高可用，zookeeper需要以叢集形态來部署，這樣隻要叢集中大部分機器是可用的(能夠容忍一定的機器故障)，那麼zookeeper本身仍然是可用的。用戶端在使用zookeeper時，需要知道叢集機器清單，通過與叢集中的某一台機器建立TCP連接配接來使用服務，用戶端使用這個TCP連結來發送請求、擷取結果、擷取監聽事件以及發送心跳包。如果這個連接配接異常斷開了，用戶端可以連接配接到另外的機器上。

架構簡圖如下所示：

用戶端的讀請求可以被叢集中的任意一台機器處理，如果讀請求在節點上注冊了監聽器，這個監聽器也是由所連接配接的zookeeper機器來處理。對于寫請求，這些請求會同時發給其他zookeeper機器并且達成一緻後，請求才會傳回成功。是以，随着zookeeper的叢集機器增多，讀請求的吞吐會提高但是寫請求的吞吐會下降。

有序性是zookeeper中非常重要的一個特性，所有的更新都是全局有序的，每個更新都有一個唯一的時間戳，這個時間戳稱為zxid(Zookeeper

Transaction Id)。而讀請求隻會相對于更新有序，也就是讀請求的傳回結果中會帶有這個zookeeper最新的zxid。

如何使用zookeeper實作分布式鎖?

在描述算法流程之前，先看下zookeeper中幾個關于節點的有趣的性質：

有序節點：假如目前有一個父節點為/lock，我們可以在這個父節點下面建立子節點;zookeeper提供了一個可選的有序特性，例如我們可以建立子節點“/lock/node-”并且指明有序，那麼zookeeper在生成子節點時會根據目前的子節點數量自動添加整數序号，也就是說如果是第一個建立的子節點，那麼生成的子節點為/lock/node-0000000000，下一個節點則為/lock/node-0000000001，依次類推。

臨時節點：用戶端可以建立一個臨時節點，在會話結束或者會話逾時後，zookeeper會自動删除該節點。

事件監聽：在讀取資料時，我們可以同時對節點設定事件監聽，當節點資料或結構變化時，zookeeper會通知用戶端。目前zookeeper有如下四種事件：1)節點建立;2)節點删除;3)節點資料修改;4)子節點變更。

下面描述使用zookeeper實作分布式鎖的算法流程，假設鎖空間的根節點為/lock：

用戶端連接配接zookeeper，并在/lock下建立臨時的且有序的子節點，第一個用戶端對應的子節點為/lock/lock-0000000000，第二個為/lock/lock-0000000001，以此類推。

用戶端擷取/lock下的子節點清單，判斷自己建立的子節點是否為目前子節點清單中序号最小的子節點，如果是則認為獲得鎖，否則監聽/lock的子節點變更消息，獲得子節點變更通知後重複此步驟直至獲得鎖;

執行業務代碼;

完成業務流程後，删除對應的子節點釋放鎖。

步驟1中建立的臨時節點能夠保證在故障的情況下鎖也能被釋放，考慮這麼個場景：假如用戶端a目前建立的子節點為序号最小的節點，獲得鎖之後用戶端所在機器當機了，用戶端沒有主動删除子節點;如果建立的是永久的節點，那麼這個鎖永遠不會釋放，導緻死鎖;由于建立的是臨時節點，用戶端當機後，過了一定時間zookeeper沒有收到用戶端的心跳包判斷會話失效，将臨時節點删除進而釋放鎖。

另外細心的朋友可能會想到，在步驟2中擷取子節點清單與設定監聽這兩步操作的原子性問題，考慮這麼個場景：用戶端a對應子節點為/lock/lock-0000000000，用戶端b對應子節點為/lock/lock-0000000001，用戶端b擷取子節點清單時發現自己不是序号最小的，但是在設定監聽器前用戶端a完成業務流程删除了子節點/lock/lock-0000000000，用戶端b設定的監聽器豈不是丢失了這個事件進而導緻永遠等待了?這個問題不存在的。因為zookeeper提供的API中設定監聽器的操作與讀操作是

原子執行的，也就是說在讀子節點清單時同時設定監聽器，保證不會丢失事件。

最後，對于這個算法有個極大的優化點：假如目前有1000個節點在等待鎖，如果獲得鎖的用戶端釋放鎖時，這1000個用戶端都會被喚醒，這種情況稱為“羊群效應”;在這種羊群效應中，zookeeper需要通知1000個用戶端，這會阻塞其他的操作，最好的情況應該隻喚醒新的最小節點對應的用戶端。應該怎麼做呢?在設定事件監聽時，每個用戶端應該對剛好在它之前的子節點設定事件監聽，例如子節點清單為/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序号為1的用戶端監聽序号為0的子節點删除消息，序号為2的監聽序号為1的子節點删除消息。

是以調整後的分布式鎖算法流程如下：

用戶端擷取/lock下的子節點清單，判斷自己建立的子節點是否為目前子節點清單中序号最小的子節點，如果是則認為獲得鎖，否則監聽剛好在自己之前一位的子節點删除消息，獲得子節點變更通知後重複此步驟直至獲得鎖;

Curator的源碼分析

雖然zookeeper原生用戶端暴露的API已經非常簡潔了，但是實作一個分布式鎖還是比較麻煩的…我們可以直接使用 curator 這個開源項目提供的zookeeper分布式鎖實作。

我們隻需要引入下面這個包(基于maven)：

<groupId>org.apache.curator</groupId>

<artifactId>curator-recipes</artifactId>

</dependency>

然後就可以用啦!代碼如下：

public static void main(String[] args) throws Exception {

//建立zookeeper的用戶端

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

CuratorFramework client = CuratorFrameworkFactory.newClient("10.21.41.181:2181,10.21.42.47:2181,10.21.49.252:2181", retryPolicy);

client.start();

//建立分布式鎖, 鎖空間的根節點路徑為/curator/lock

InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");

mutex.acquire();

//獲得了鎖, 進行業務流程

System.out.println("Enter mutex");

//完成業務流程, 釋放鎖

mutex.release();

//關閉用戶端

client.close();

}

可以看到關鍵的核心操作就隻有mutex.acquire()和mutex.release()，簡直太友善了!

下面來分析下擷取鎖的源碼實作。acquire的方法如下：

* 擷取鎖，當鎖被占用時會阻塞等待，這個操作支援同線程的可重入（也就是重複擷取鎖），acquire的次數需要與release的次數相同。

* @throws Exception ZK errors, connection interruptions

@Override

public void acquire() throws Exception

{

if ( !internalLock(-1, null) )

{

throw new IOException("Lost connection while trying to acquire lock: " + basePath);

}

這裡有個地方需要注意，當與zookeeper通信存在異常時，acquire會直接抛出異常，需要使用者自身做重試政策。代碼中調用了internalLock(-1, null)，參數表明在鎖被占用時永久阻塞等待。internalLock的代碼如下：

private boolean internalLock(long time, TimeUnit unit) throws Exception

//這裡處理同線程的可重入性，如果已經獲得鎖，那麼隻是在對應的資料結構中增加acquire的次數統計，直接傳回成功

Thread currentThread = Thread.currentThread();

LockData lockData = threadData.get(currentThread);

if ( lockData != null )

// re-entering

lockData.lockCount.incrementAndGet();

return true;

//這裡才真正去zookeeper中擷取鎖

String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());

if ( lockPath != null )

//獲得鎖之後，記錄目前的線程獲得鎖的資訊，在重入時隻需在LockData中增加次數統計即可

LockData newLockData = new LockData(currentThread, lockPath);

threadData.put(currentThread, newLockData);

//在阻塞傳回時仍然擷取不到鎖，這裡上下文的處理隐含的意思為zookeeper通信異常

return false;

代碼中增加了具體注釋，不做展開。看下zookeeper擷取鎖的具體實作：

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception

//參數初始化，此處省略

//...

//自旋擷取鎖

while ( !isDone )

isDone = true;

try

{

//在鎖空間下建立臨時且有序的子節點

ourPath = driver.createsTheLock(client, path, localLockNodeBytes);

//判斷是否獲得鎖（子節點序号最小），獲得鎖則直接傳回，否則阻塞等待前一個子節點删除通知

hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);

}

catch ( KeeperException.NoNodeException e )

//對于NoNodeException，代碼中確定了隻有發生session過期才會在這裡抛出NoNodeException，是以這裡根據重試政策進行重試

if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )

{

isDone = false;

}

else

throw e;

//如果獲得鎖則傳回該子節點的路徑

if ( hasTheLock )

return ourPath;

return null;

上面代碼中主要有兩步操作：

driver.createsTheLock：建立臨時且有序的子節點，裡面實作比較簡單不做展開，主要關注幾種節點的模式：1)PERSISTENT(永久);2)PERSISTENT_SEQUENTIAL(永久且有序);3)EPHEMERAL(臨時);4)EPHEMERAL_SEQUENTIAL(臨時且有序)。

internalLockLoop：阻塞等待直到獲得鎖。

看下internalLockLoop是怎麼判斷鎖以及阻塞等待的，這裡删除了一些無關代碼，隻保留主流程：

//自旋直至獲得鎖

while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )

//擷取所有的子節點清單，并且按序号從小到大排序

List<String> children = getSortedChildren();

//根據序号判斷目前子節點是否為最小子節點

String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash

PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);

if ( predicateResults.getsTheLock() )

//如果為最小子節點則認為獲得鎖

haveTheLock = true;

else

//否則擷取前一個子節點

String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();

//這裡使用對象螢幕做線程同步，當擷取不到鎖時監聽前一個子節點删除消息并且進行wait()，目前一個子節點删除（也就是鎖釋放）時，回調會通過notifyAll喚醒此線程，此線程繼續自旋判斷是否獲得鎖

synchronized(this)

try

//這裡使用getData()接口而不是checkExists()是因為，如果前一個子節點已經被删除了那麼會抛出異常而且不會設定事件監聽器，而checkExists雖然也可以擷取到節點是否存在的資訊但是同時設定了監聽器，這個監聽器其實永遠不會觸發，對于zookeeper來說屬于資源洩露

client.getData().usingWatcher(watcher).forPath(previousSequencePath);

//如果設定了阻塞等待的時間

if ( millisToWait != null )

{

millisToWait -= (System.currentTimeMillis() - startMillis);

startMillis = System.currentTimeMillis();

if ( millisToWait <= 0 )

{

doDelete = true; // 等待時間到達，删除對應的子節點

break;

}

//等待相應的時間

wait(millisToWait);

}

else

//永遠等待

wait();

catch ( KeeperException.NoNodeException e )

//上面使用getData來設定監聽器時，如果前一個子節點已經被删除那麼會抛出NoNodeException，隻需要自旋一次即可，無需額外處理

具體邏輯見注釋，不再贅述。代碼中設定的事件監聽器，在事件發生回調時隻是簡單的notifyAll喚醒目前線程以重新自旋判斷，比較簡單不再展開。

以上。

本文作者：佚名

來源：51CTO

基于Zookeeper的分布式鎖

繼續閱讀

Codeforces 1417 D. Make Them Equal(思維+構造)

查找算法之二分查找查找算法之二分查找

查找算法學習之二分查找（Python版本）——BinarySearch

CQ V1.0分詞bates(基于雙數組tire樹)—應該是目前最快的中文分詞算法

Command Network(POJ 3164)---定根最小樹形圖模闆題題目描述輸入格式輸出格式輸入樣例輸出樣例分析源程式

開源低帶寬語音編解碼器

241 Different Ways to Add Parentheses（C代碼版）

【趨高機器視覺】機器視覺技術原了解析及解決方案

CSMA/CD1． CSMA/CD的概述2． CSMA 的工作原理3． CSMA/CD控制規程及特點4． CSMA/CD協定5． CSMA/CD的優點6．結束語

極大似然法(ML)與最大期望法(EM)

C++ 第十五周報告1--《冒泡法排序》

筆試面試題目：滑動視窗(二)

資料結構與算法（27）——排序（二）

Dijkstra--簡易版（最短路徑）

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

hdu7108哈希