什么是Zookeeper？

Zookeeper（業(yè)界簡稱zk）是一種提供配置管理、分布式協(xié)同以及命名的中心化服務(wù)，這些提供的功能都是分布式系統(tǒng)中非常底層且必不可少的基本功能，但是如果自己實(shí)現(xiàn)這些功能而且要達(dá)到高吞吐、低延遲同時還要保持一致性和可用性，實(shí)際上非常困難。因此zookeeper提供了這些功能，開發(fā)者在zookeeper之上構(gòu)建自己的各種分布式系統(tǒng)。

雖然zookeeper的實(shí)現(xiàn)比較復(fù)雜，但是它提供的模型抽象卻是非常簡單的。Zookeeper提供一個多層級的節(jié)點(diǎn)命名空間（節(jié)點(diǎn)稱為znode），每個節(jié)點(diǎn)都用一個以斜杠（/）分隔的路徑表示，而且每個節(jié)點(diǎn)都有父節(jié)點(diǎn)（根節(jié)點(diǎn)除外），非常類似于文件系統(tǒng)。例如，/foo/doo這個表示一個znode，它的父節(jié)點(diǎn)為/foo，父父節(jié)點(diǎn)為/，而/為根節(jié)點(diǎn)沒有父節(jié)點(diǎn)。與文件系統(tǒng)不同的是，這些節(jié)點(diǎn)都可以設(shè)置關(guān)聯(lián)的數(shù)據(jù)，而文件系統(tǒng)中只有文件節(jié)點(diǎn)可以存放數(shù)據(jù)而目錄節(jié)點(diǎn)不行。Zookeeper為了保證高吞吐和低延遲，在內(nèi)存中維護(hù)了這個樹狀的目錄結(jié)構(gòu)，這種特性使得Zookeeper不能用于存放大量的數(shù)據(jù)，每個節(jié)點(diǎn)的存放數(shù)據(jù)上限為1M。

而為了保證高可用，zookeeper需要以集群形態(tài)來部署，這樣只要集群中大部分機(jī)器是可用的（能夠容忍一定的機(jī)器故障），那么zookeeper本身仍然是可用的。客戶端在使用zookeeper時，需要知道集群機(jī)器列表，通過與集群中的某一臺機(jī)器建立TCP連接來使用服務(wù)，客戶端使用這個TCP鏈接來發(fā)送請求、獲取結(jié)果、獲取監(jiān)聽事件以及發(fā)送心跳包。如果這個連接異常斷開了，客戶端可以連接到另外的機(jī)器上。

架構(gòu)簡圖如下所示：

zookeeper

客戶端的讀請求可以被集群中的任意一臺機(jī)器處理，如果讀請求在節(jié)點(diǎn)上注冊了監(jiān)聽器，這個監(jiān)聽器也是由所連接的zookeeper機(jī)器來處理。對于寫請求，這些請求會同時發(fā)給其他zookeeper機(jī)器并且達(dá)成一致后，請求才會返回成功。因此，隨著zookeeper的集群機(jī)器增多，讀請求的吞吐會提高但是寫請求的吞吐會下降。

有序性是zookeeper中非常重要的一個特性，所有的更新都是全局有序的，每個更新都有一個唯一的時間戳，這個時間戳稱為zxid（Zookeeper Transaction Id）。而讀請求只會相對于更新有序，也就是讀請求的返回結(jié)果中會帶有這個zookeeper最新的zxid。

如何使用zookeeper實(shí)現(xiàn)分布式鎖？

在描述算法流程之前，先看下zookeeper中幾個關(guān)于節(jié)點(diǎn)的有趣的性質(zhì)：

• 有序節(jié)點(diǎn)：假如當(dāng)前有一個父節(jié)點(diǎn)為/lock，我們可以在這個父節(jié)點(diǎn)下面創(chuàng)建子節(jié)點(diǎn)；zookeeper提供了一個可選的有序特性，例如我們可以創(chuàng)建子節(jié)點(diǎn)“/lock/node-”并且指明有序，那么zookeeper在生成子節(jié)點(diǎn)時會根據(jù)當(dāng)前的子節(jié)點(diǎn)數(shù)量自動添加整數(shù)序號，也就是說如果是第一個創(chuàng)建的子節(jié)點(diǎn)，那么生成的子節(jié)點(diǎn)為/lock/node-0000000000，下一個節(jié)點(diǎn)則為/lock/node-0000000001，依次類推。

• 臨時節(jié)點(diǎn)：客戶端可以建立一個臨時節(jié)點(diǎn)，在會話結(jié)束或者會話超時后，zookeeper會自動刪除該節(jié)點(diǎn)。

• 事件監(jiān)聽：在讀取數(shù)據(jù)時，我們可以同時對節(jié)點(diǎn)設(shè)置事件監(jiān)聽，當(dāng)節(jié)點(diǎn)數(shù)據(jù)或結(jié)構(gòu)變化時，zookeeper會通知客戶端。當(dāng)前zookeeper有如下四種事件：1）節(jié)點(diǎn)創(chuàng)建；2）節(jié)點(diǎn)刪除；3）節(jié)點(diǎn)數(shù)據(jù)修改；4）子節(jié)點(diǎn)變更。

下面描述使用zookeeper實(shí)現(xiàn)分布式鎖的算法流程，假設(shè)鎖空間的根節(jié)點(diǎn)為/lock：

客戶端連接zookeeper，并在/lock下創(chuàng)建臨時的且有序的子節(jié)點(diǎn)，第一個客戶端對應(yīng)的子節(jié)點(diǎn)為/lock/lock-0000000000，第二個為/lock/lock-0000000001，以此類推。

客戶端獲取/lock下的子節(jié)點(diǎn)列表，判斷自己創(chuàng)建的子節(jié)點(diǎn)是否為當(dāng)前子節(jié)點(diǎn)列表中序號最小的子節(jié)點(diǎn)，如果是則認(rèn)為獲得鎖，否則監(jiān)聽/lock的子節(jié)點(diǎn)變更消息，獲得子節(jié)點(diǎn)變更通知后重復(fù)此步驟直至獲得鎖；

執(zhí)行業(yè)務(wù)代碼；

完成業(yè)務(wù)流程后，刪除對應(yīng)的子節(jié)點(diǎn)釋放鎖。

步驟1中創(chuàng)建的臨時節(jié)點(diǎn)能夠保證在故障的情況下鎖也能被釋放，考慮這么個場景：假如客戶端a當(dāng)前創(chuàng)建的子節(jié)點(diǎn)為序號最小的節(jié)點(diǎn)，獲得鎖之后客戶端所在機(jī)器宕機(jī)了，客戶端沒有主動刪除子節(jié)點(diǎn)；如果創(chuàng)建的是永久的節(jié)點(diǎn)，那么這個鎖永遠(yuǎn)不會釋放，導(dǎo)致死鎖；由于創(chuàng)建的是臨時節(jié)點(diǎn)，客戶端宕機(jī)后，過了一定時間zookeeper沒有收到客戶端的心跳包判斷會話失效，將臨時節(jié)點(diǎn)刪除從而釋放鎖。

另外細(xì)心的朋友可能會想到，在步驟2中獲取子節(jié)點(diǎn)列表與設(shè)置監(jiān)聽這兩步操作的原子性問題，考慮這么個場景：客戶端a對應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000000，客戶端b對應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000001，客戶端b獲取子節(jié)點(diǎn)列表時發(fā)現(xiàn)自己不是序號最小的，但是在設(shè)置監(jiān)聽器前客戶端a完成業(yè)務(wù)流程刪除了子節(jié)點(diǎn)/lock/lock-0000000000，客戶端b設(shè)置的監(jiān)聽器豈不是丟失了這個事件從而導(dǎo)致永遠(yuǎn)等待了？這個問題不存在的。因?yàn)閦ookeeper提供的API中設(shè)置監(jiān)聽器的操作與讀操作是原子執(zhí)行的，也就是說在讀子節(jié)點(diǎn)列表時同時設(shè)置監(jiān)聽器，保證不會丟失事件。

最后，對于這個算法有個極大的優(yōu)化點(diǎn)：假如當(dāng)前有1000個節(jié)點(diǎn)在等待鎖，如果獲得鎖的客戶端釋放鎖時，這1000個客戶端都會被喚醒，這種情況稱為“羊群效應(yīng)”；在這種羊群效應(yīng)中，zookeeper需要通知1000個客戶端，這會阻塞其他的操作，最好的情況應(yīng)該只喚醒新的最小節(jié)點(diǎn)對應(yīng)的客戶端。應(yīng)該怎么做呢？在設(shè)置事件監(jiān)聽時，每個客戶端應(yīng)該對剛好在它之前的子節(jié)點(diǎn)設(shè)置事件監(jiān)聽，例如子節(jié)點(diǎn)列表為/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序號為1的客戶端監(jiān)聽序號為0的子節(jié)點(diǎn)刪除消息，序號為2的監(jiān)聽序號為1的子節(jié)點(diǎn)刪除消息。

zookeeper學(xué)習(xí)中

所以調(diào)整后的分布式鎖算法流程如下：

• 客戶端連接zookeeper，并在/lock下創(chuàng)建臨時的且有序的子節(jié)點(diǎn)，第一個客戶端對應(yīng)的子節(jié)點(diǎn)為/lock/lock-0000000000，第二個為/lock/lock-0000000001，以此類推；

• 客戶端獲取/lock下的子節(jié)點(diǎn)列表，判斷自己創(chuàng)建的子節(jié)點(diǎn)是否為當(dāng)前子節(jié)點(diǎn)列表中序號最小的子節(jié)點(diǎn)，如果是則認(rèn)為獲得鎖，否則監(jiān)聽剛好在自己之前一位的子節(jié)點(diǎn)刪除消息，獲得子節(jié)點(diǎn)變更通知后重復(fù)此步驟直至獲得鎖；

• 執(zhí)行業(yè)務(wù)代碼；

• 完成業(yè)務(wù)流程后，刪除對應(yīng)的子節(jié)點(diǎn)釋放鎖。

Curator的源碼分析

雖然zookeeper原生客戶端暴露的API已經(jīng)非常簡潔了，但是實(shí)現(xiàn)一個分布式鎖還是比較麻煩的…我們可以直接使用curator這個開源項(xiàng)目提供的zookeeper分布式鎖實(shí)現(xiàn)。

我們只需要引入下面這個包（基于maven）：

1 <dependency> 2 3 <groupId>org.apache.curator</groupId> 4 5 <artifactId>curator-recipes</artifactId> 6 7 <version>4.0.0</version> 8 9 </dependency>

?

然后就可以用啦！代碼如下：

1 public static void main(String[] args) throws Exception { 2 3 //創(chuàng)建zookeeper的客戶端 4 5 RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); 6 7 CuratorFramework client = CuratorFrameworkFactory.newClient("10.21.41.181:2181,10.21.42.47:2181,10.21.49.252:2181", retryPolicy); 8 9 client.start(); 10 11 //創(chuàng)建分布式鎖, 鎖空間的根節(jié)點(diǎn)路徑為/curator/lock 12 13 InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock"); 14 15 mutex.acquire(); 16 17 //獲得了鎖, 進(jìn)行業(yè)務(wù)流程 18 19 System.out.println("Enter mutex"); 20 21 //完成業(yè)務(wù)流程, 釋放鎖 22 23 mutex.release(); 24 25 //關(guān)閉客戶端 26 27 client.close(); 28 29 }

?

　　

可以看到關(guān)鍵的核心操作就只有mutex.acquire()和mutex.release()，簡直太方便了！

下面來分析下獲取鎖的源碼實(shí)現(xiàn)。acquire的方法如下：

1 /* 2 3 * 獲取鎖，當(dāng)鎖被占用時會阻塞等待，這個操作支持同線程的可重入（也就是重復(fù)獲取鎖），acquire的次數(shù)需要與release的次數(shù)相同。 4 5 * @throws Exception ZK errors, connection interruptions 6 7 */ 8 9 @Override 10 11 public void acquire() throws Exception 12 13 { 14 15 if ( !internalLock(-1, null) ) 16 17 { 18 19 throw new IOException("Lost connection while trying to acquire lock: " + basePath); 20 21 } 22 23 } View Code

?

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

private boolean internalLock(long time, TimeUnit unit) throws Exception{//這里處理同線程的可重入性，如果已經(jīng)獲得鎖，那么只是在對應(yīng)的數(shù)據(jù)結(jié)構(gòu)中增加acquire的次數(shù)統(tǒng)計(jì)，直接返回成功 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 ){//獲得鎖之后，記錄當(dāng)前的線程獲得鎖的信息，在重入時只需在LockData中增加次數(shù)統(tǒng)計(jì)即可 LockData newLockData = new LockData(currentThread, lockPath);threadData.put(currentThread, newLockData);return true;}//在阻塞返回時仍然獲取不到鎖，這里上下文的處理隱含的意思為zookeeper通信異常return false;} View Code

?

代碼中增加了具體注釋，不做展開。看下zookeeper獲取鎖的具體實(shí)現(xiàn)：

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception{//參數(shù)初始化，此處省略//...//自旋獲取鎖while ( !isDone ){isDone = true;try{//在鎖空間下創(chuàng)建臨時且有序的子節(jié)點(diǎn) ourPath = driver.createsTheLock(client, path, localLockNodeBytes);//判斷是否獲得鎖（子節(jié)點(diǎn)序號最小），獲得鎖則直接返回，否則阻塞等待前一個子節(jié)點(diǎn)刪除通知 hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);}catch ( KeeperException.NoNodeException e ){//對于NoNodeException，代碼中確保了只有發(fā)生session過期才會在這里拋出NoNodeException，因此這里根據(jù)重試策略進(jìn)行重試if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) ){isDone = false;}else{throw e;}}}//如果獲得鎖則返回該子節(jié)點(diǎn)的路徑if ( hasTheLock ){return ourPath;}return null;} View Code

?

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

• driver.createsTheLock：創(chuàng)建臨時且有序的子節(jié)點(diǎn)，里面實(shí)現(xiàn)比較簡單不做展開，主要關(guān)注幾種節(jié)點(diǎn)的模式：1）PERSISTENT（永久）；2）PERSISTENT_SEQUENTIAL（永久且有序）；3）EPHEMERAL（臨時）；4）EPHEMERAL_SEQUENTIAL（臨時且有序）。

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

看下internalLockLoop是怎么判斷鎖以及阻塞等待的，這里刪除了一些無關(guān)代碼，只保留主流程：

//自旋直至獲得鎖while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ){//獲取所有的子節(jié)點(diǎn)列表，并且按序號從小到大排序 List<String> children = getSortedChildren();//根據(jù)序號判斷當(dāng)前子節(jié)點(diǎn)是否為最小子節(jié)點(diǎn) String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);if ( predicateResults.getsTheLock() ){//如果為最小子節(jié)點(diǎn)則認(rèn)為獲得鎖 haveTheLock = true;}else{//否則獲取前一個子節(jié)點(diǎn) String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();//這里使用對象監(jiān)視器做線程同步，當(dāng)獲取不到鎖時監(jiān)聽前一個子節(jié)點(diǎn)刪除消息并且進(jìn)行wait()，當(dāng)前一個子節(jié)點(diǎn)刪除（也就是鎖釋放）時，回調(diào)會通過notifyAll喚醒此線程，此線程繼續(xù)自旋判斷是否獲得鎖synchronized(this){try{//這里使用getData()接口而不是checkExists()是因?yàn)?#xff0c;如果前一個子節(jié)點(diǎn)已經(jīng)被刪除了那么會拋出異常而且不會設(shè)置事件監(jiān)聽器，而checkExists雖然也可以獲取到節(jié)點(diǎn)是否存在的信息但是同時設(shè)置了監(jiān)聽器，這個監(jiān)聽器其實(shí)永遠(yuǎn)不會觸發(fā)，對于zookeeper來說屬于資源泄露 client.getData().usingWatcher(watcher).forPath(previousSequencePath);//如果設(shè)置了阻塞等待的時間if ( millisToWait != null ){millisToWait -= (System.currentTimeMillis() - startMillis);startMillis = System.currentTimeMillis();if ( millisToWait <= 0 ){doDelete = true; // 等待時間到達(dá)，刪除對應(yīng)的子節(jié)點(diǎn)break;}//等待相應(yīng)的時間 wait(millisToWait);}else{//永遠(yuǎn)等待 wait();}}catch ( KeeperException.NoNodeException e ){//上面使用getData來設(shè)置監(jiān)聽器時，如果前一個子節(jié)點(diǎn)已經(jīng)被刪除那么會拋出NoNodeException，只需要自旋一次即可，無需額外處理 }}}} View Code

?

轉(zhuǎn)載自：https://blog.csdn.net/qiangcuo6087/article/details/79067136

轉(zhuǎn)載于:https://www.cnblogs.com/itplay/p/9510640.html

與50位技術(shù)專家面對面20年技術(shù)見證，附贈技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的分布式锁（一） Zookeeper分布式锁的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。