建站
咨詢(xún)
售后
建站咨詢(xún)
新聞中心
基于ZooKeeper的分布式鎖
作者：佚名 2017-10-24 11:28:23
開(kāi)發(fā)
前端
分布式 實(shí)現(xiàn)分布式鎖目前有三種流行方案，分別為基于數(shù)據(jù)庫(kù)、Redis、Zookeeper的方案，其中前兩種方案網(wǎng)絡(luò)上有很多資料可以參考，本文不做展開(kāi)。我們來(lái)看下使用Zookeeper如何實(shí)現(xiàn)分布式鎖。
從網(wǎng)站建設(shè)到定制行業(yè)解決方案，為提供網(wǎng)站制作、成都網(wǎng)站制作服務(wù)體系，各種行業(yè)企業(yè)客戶(hù)提供網(wǎng)站建設(shè)解決方案，助力業(yè)務(wù)快速發(fā)展。創(chuàng)新互聯(lián)公司將不斷加快創(chuàng)新步伐，提供優(yōu)質(zhì)的建站服務(wù)。
這篇文章只需要你10分鐘的時(shí)間。
實(shí)現(xiàn)分布式鎖目前有三種流行方案，分別為基于數(shù)據(jù)庫(kù)、Redis、Zookeeper的方案，其中前兩種方案網(wǎng)絡(luò)上有很多資料可以參考，本文不做展開(kāi)。我們來(lái)看下使用Zookeeper如何實(shí)現(xiàn)分布式鎖。
什么是Zookeeper?
Zookeeper(業(yè)界簡(jiǎn)稱(chēng)zk)是一種提供配置管理、分布式協(xié)同以及命名的中心化服務(wù)，這些提供的功能都是分布式系統(tǒng)中非常底層且必不可少的基本功能，但是如果自己實(shí)現(xiàn)這些功能而且要達(dá)到高吞吐、低延遲同時(shí)還要保持一致性和可用性，實(shí)際上非常困難。因此zookeeper提供了這些功能，開(kāi)發(fā)者在zookeeper之上構(gòu)建自己的各種分布式系統(tǒng)。
雖然zookeeper的實(shí)現(xiàn)比較復(fù)雜，但是它提供的模型抽象卻是非常簡(jiǎn)單的。Zookeeper提供一個(gè)多層級(jí)的節(jié)點(diǎn)命名空間(節(jié)點(diǎn)稱(chēng)為znode)，每個(gè)節(jié)點(diǎn)都用一個(gè)以斜杠(/)分隔的路徑表示，而且每個(gè)節(jié)點(diǎn)都有父節(jié)點(diǎn)(根節(jié)點(diǎn)除外)，非常類(lèi)似于文件系統(tǒng)。例如，/foo/doo這個(gè)表示一個(gè)znode，它的父節(jié)點(diǎn)為/foo，父父節(jié)點(diǎn)為/，而/為根節(jié)點(diǎn)沒(méi)有父節(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ù)了這個(gè)樹(shù)狀的目錄結(jié)構(gòu)，這種特性使得Zookeeper不能用于存放大量的數(shù)據(jù)，每個(gè)節(jié)點(diǎn)的存放數(shù)據(jù)上限為1M。
而為了保證高可用，zookeeper需要以集群形態(tài)來(lái)部署，這樣只要集群中大部分機(jī)器是可用的(能夠容忍一定的機(jī)器故障)，那么zookeeper本身仍然是可用的。客戶(hù)端在使用zookeeper時(shí)，需要知道集群機(jī)器列表，通過(guò)與集群中的某一臺(tái)機(jī)器建立TCP連接來(lái)使用服務(wù)，客戶(hù)端使用這個(gè)TCP鏈接來(lái)發(fā)送請(qǐng)求、獲取結(jié)果、獲取監(jiān)聽(tīng)事件以及發(fā)送心跳包。如果這個(gè)連接異常斷開(kāi)了，客戶(hù)端可以連接到另外的機(jī)器上。
架構(gòu)簡(jiǎn)圖如下所示：
客戶(hù)端的讀請(qǐng)求可以被集群中的任意一臺(tái)機(jī)器處理，如果讀請(qǐng)求在節(jié)點(diǎn)上注冊(cè)了監(jiān)聽(tīng)器，這個(gè)監(jiān)聽(tīng)器也是由所連接的zookeeper機(jī)器來(lái)處理。對(duì)于寫(xiě)請(qǐng)求，這些請(qǐng)求會(huì)同時(shí)發(fā)給其他zookeeper機(jī)器并且達(dá)成一致后，請(qǐng)求才會(huì)返回成功。因此，隨著zookeeper的集群機(jī)器增多，讀請(qǐng)求的吞吐會(huì)提高但是寫(xiě)請(qǐng)求的吞吐會(huì)下降。
有序性是zookeeper中非常重要的一個(gè)特性，所有的更新都是全局有序的，每個(gè)更新都有一個(gè)唯一的時(shí)間戳，這個(gè)時(shí)間戳稱(chēng)為zxid(Zookeeper Transaction Id)。而讀請(qǐng)求只會(huì)相對(duì)于更新有序，也就是讀請(qǐng)求的返回結(jié)果中會(huì)帶有這個(gè)zookeeper***的zxid。
如何使用zookeeper實(shí)現(xiàn)分布式鎖?
在描述算法流程之前，先看下zookeeper中幾個(gè)關(guān)于節(jié)點(diǎn)的有趣的性質(zhì)：
- 有序節(jié)點(diǎn)：假如當(dāng)前有一個(gè)父節(jié)點(diǎn)為/lock，我們可以在這個(gè)父節(jié)點(diǎn)下面創(chuàng)建子節(jié)點(diǎn);zookeeper提供了一個(gè)可選的有序特性，例如我們可以創(chuàng)建子節(jié)點(diǎn)“/lock/node-”并且指明有序，那么zookeeper在生成子節(jié)點(diǎn)時(shí)會(huì)根據(jù)當(dāng)前的子節(jié)點(diǎn)數(shù)量自動(dòng)添加整數(shù)序號(hào)，也就是說(shuō)如果是***個(gè)創(chuàng)建的子節(jié)點(diǎn)，那么生成的子節(jié)點(diǎn)為/lock/node-0000000000，下一個(gè)節(jié)點(diǎn)則為/lock/node-0000000001，依次類(lèi)推。
- 臨時(shí)節(jié)點(diǎn)：客戶(hù)端可以建立一個(gè)臨時(shí)節(jié)點(diǎn)，在會(huì)話結(jié)束或者會(huì)話超時(shí)后，zookeeper會(huì)自動(dòng)刪除該節(jié)點(diǎn)。
- 事件監(jiān)聽(tīng)：在讀取數(shù)據(jù)時(shí)，我們可以同時(shí)對(duì)節(jié)點(diǎn)設(shè)置事件監(jiān)聽(tīng)，當(dāng)節(jié)點(diǎn)數(shù)據(jù)或結(jié)構(gòu)變化時(shí)，zookeeper會(huì)通知客戶(hù)端。當(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：
- 客戶(hù)端連接zookeeper，并在/lock下創(chuàng)建 臨時(shí)的 且 有序的 子節(jié)點(diǎn)，***個(gè)客戶(hù)端對(duì)應(yīng)的子節(jié)點(diǎn)為/lock/lock-0000000000，第二個(gè)為/lock/lock-0000000001，以此類(lèi)推。
- 客戶(hù)端獲取/lock下的子節(jié)點(diǎn)列表，判斷自己創(chuàng)建的子節(jié)點(diǎn)是否為當(dāng)前子節(jié)點(diǎn)列表中 序號(hào)最小 的子節(jié)點(diǎn)，如果是則認(rèn)為獲得鎖，否則監(jiān)聽(tīng)/lock的子節(jié)點(diǎn)變更消息，獲得子節(jié)點(diǎn)變更通知后重復(fù)此步驟直至獲得鎖;
- 執(zhí)行業(yè)務(wù)代碼;
- 完成業(yè)務(wù)流程后，刪除對(duì)應(yīng)的子節(jié)點(diǎn)釋放鎖。
步驟1中創(chuàng)建的臨時(shí)節(jié)點(diǎn)能夠保證在故障的情況下鎖也能被釋放，考慮這么個(gè)場(chǎng)景：假如客戶(hù)端a當(dāng)前創(chuàng)建的子節(jié)點(diǎn)為序號(hào)最小的節(jié)點(diǎn)，獲得鎖之后客戶(hù)端所在機(jī)器宕機(jī)了，客戶(hù)端沒(méi)有主動(dòng)刪除子節(jié)點(diǎn);如果創(chuàng)建的是***的節(jié)點(diǎn)，那么這個(gè)鎖永遠(yuǎn)不會(huì)釋放，導(dǎo)致死鎖;由于創(chuàng)建的是臨時(shí)節(jié)點(diǎn)，客戶(hù)端宕機(jī)后，過(guò)了一定時(shí)間zookeeper沒(méi)有收到客戶(hù)端的心跳包判斷會(huì)話失效，將臨時(shí)節(jié)點(diǎn)刪除從而釋放鎖。
另外細(xì)心的朋友可能會(huì)想到，在步驟2中獲取子節(jié)點(diǎn)列表與設(shè)置監(jiān)聽(tīng)這兩步操作的原子性問(wèn)題，考慮這么個(gè)場(chǎng)景：客戶(hù)端a對(duì)應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000000，客戶(hù)端b對(duì)應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000001，客戶(hù)端b獲取子節(jié)點(diǎn)列表時(shí)發(fā)現(xiàn)自己不是序號(hào)最小的，但是在設(shè)置監(jiān)聽(tīng)器前客戶(hù)端a完成業(yè)務(wù)流程刪除了子節(jié)點(diǎn)/lock/lock-0000000000，客戶(hù)端b設(shè)置的監(jiān)聽(tīng)器豈不是丟失了這個(gè)事件從而導(dǎo)致永遠(yuǎn)等待了?這個(gè)問(wèn)題不存在的。因?yàn)閦ookeeper提供的API中設(shè)置監(jiān)聽(tīng)器的操作與讀操作是 原子執(zhí)行 的，也就是說(shuō)在讀子節(jié)點(diǎn)列表時(shí)同時(shí)設(shè)置監(jiān)聽(tīng)器，保證不會(huì)丟失事件。
***，對(duì)于這個(gè)算法有個(gè)極大的優(yōu)化點(diǎn)：假如當(dāng)前有1000個(gè)節(jié)點(diǎn)在等待鎖，如果獲得鎖的客戶(hù)端釋放鎖時(shí)，這1000個(gè)客戶(hù)端都會(huì)被喚醒，這種情況稱(chēng)為“羊群效應(yīng)”;在這種羊群效應(yīng)中，zookeeper需要通知1000個(gè)客戶(hù)端，這會(huì)阻塞其他的操作，***的情況應(yīng)該只喚醒新的最小節(jié)點(diǎn)對(duì)應(yīng)的客戶(hù)端。應(yīng)該怎么做呢?在設(shè)置事件監(jiān)聽(tīng)時(shí)，每個(gè)客戶(hù)端應(yīng)該對(duì)剛好在它之前的子節(jié)點(diǎn)設(shè)置事件監(jiān)聽(tīng)，例如子節(jié)點(diǎn)列表為/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序號(hào)為1的客戶(hù)端監(jiān)聽(tīng)序號(hào)為0的子節(jié)點(diǎn)刪除消息，序號(hào)為2的監(jiān)聽(tīng)序號(hào)為1的子節(jié)點(diǎn)刪除消息。
所以調(diào)整后的分布式鎖算法流程如下：
- 客戶(hù)端連接zookeeper，并在/lock下創(chuàng)建 臨時(shí)的 且 有序的 子節(jié)點(diǎn)，***個(gè)客戶(hù)端對(duì)應(yīng)的子節(jié)點(diǎn)為/lock/lock-0000000000，第二個(gè)為/lock/lock-0000000001，以此類(lèi)推。
- 客戶(hù)端獲取/lock下的子節(jié)點(diǎn)列表，判斷自己創(chuàng)建的子節(jié)點(diǎn)是否為當(dāng)前子節(jié)點(diǎn)列表中 序號(hào)最小 的子節(jié)點(diǎn)，如果是則認(rèn)為獲得鎖，否則 監(jiān)聽(tīng)剛好在自己之前一位的子節(jié)點(diǎn)刪除消息 ，獲得子節(jié)點(diǎn)變更通知后重復(fù)此步驟直至獲得鎖;
- 執(zhí)行業(yè)務(wù)代碼;
- 完成業(yè)務(wù)流程后，刪除對(duì)應(yīng)的子節(jié)點(diǎn)釋放鎖。
Curator的源碼分析
雖然zookeeper原生客戶(hù)端暴露的API已經(jīng)非常簡(jiǎn)潔了，但是實(shí)現(xiàn)一個(gè)分布式鎖還是比較麻煩的…我們可以直接使用 curator 這個(gè)開(kāi)源項(xiàng)目提供的zookeeper分布式鎖實(shí)現(xiàn)。
我們只需要引入下面這個(gè)包(基于maven)：
org.apache.curator curator-recipes 4.0.0
然后就可以用啦!代碼如下：
- public static void main(String[] args) throws Exception {
- //創(chuàng)建zookeeper的客戶(hù)端
- 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();
- //創(chuàng)建分布式鎖, 鎖空間的根節(jié)點(diǎn)路徑為/curator/lock
- InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
- mutex.acquire();
- //獲得了鎖, 進(jìn)行業(yè)務(wù)流程
- System.out.println("Enter mutex");
- //完成業(yè)務(wù)流程, 釋放鎖
- mutex.release();
- //關(guān)閉客戶(hù)端
- client.close();
- }
可以看到關(guān)鍵的核心操作就只有mutex.acquire()和mutex.release()，簡(jiǎn)直太方便了!
下面來(lái)分析下獲取鎖的源碼實(shí)現(xiàn)。acquire的方法如下：
- /*
- * 獲取鎖，當(dāng)鎖被占用時(shí)會(huì)阻塞等待，這個(gè)操作支持同線程的可重入（也就是重復(fù)獲取鎖），acquire的次數(shù)需要與release的次數(shù)相同。
- * @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);
- }
- }
這里有個(gè)地方需要注意，當(dāng)與zookeeper通信存在異常時(shí)，acquire會(huì)直接拋出異常，需要使用者自身做重試策略。代碼中調(diào)用了internalLock(-1, null)，參數(shù)表明在鎖被占用時(shí)***阻塞等待。internalLock的代碼如下：
- private boolean internalLock(long time, TimeUnit unit) throws Exception
- {
- //這里處理同線程的可重入性，如果已經(jīng)獲得鎖，那么只是在對(duì)應(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)前的線程獲得鎖的信息，在重入時(shí)只需在LockData中增加次數(shù)統(tǒng)計(jì)即可
- LockData newLockData = new LockData(currentThread, lockPath);
- threadData.put(currentThread, newLockData);
- return true;
- }
- //在阻塞返回時(shí)仍然獲取不到鎖，這里上下文的處理隱含的意思為zookeeper通信異常
- return false;
- }
代碼中增加了具體注釋?zhuān)蛔稣归_(kāi)?？聪聑ookeeper獲取鎖的具體實(shí)現(xiàn)：
- String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
- {
- //參數(shù)初始化，此處省略
- //...
- //自旋獲取鎖
- while ( !isDone )
- {
- isDone = true;
- try
- {
- //在鎖空間下創(chuàng)建臨時(shí)且有序的子節(jié)點(diǎn)
- ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
- //判斷是否獲得鎖（子節(jié)點(diǎn)序號(hào)最小），獲得鎖則直接返回，否則阻塞等待前一個(gè)子節(jié)點(diǎn)刪除通知
- hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
- }
- catch ( KeeperException.NoNodeException e )
- {
- //對(duì)于NoNodeException，代碼中確保了只有發(fā)生session過(guò)期才會(huì)在這里拋出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;
- }
上面代碼中主要有兩步操作：
- driver.createsTheLock：創(chuàng)建臨時(shí)且有序的子節(jié)點(diǎn)，里面實(shí)現(xiàn)比較簡(jiǎn)單不做展開(kāi)，主要關(guān)注幾種節(jié)點(diǎn)的模式：1)PERSISTENT(***);2)PERSISTENT_SEQUENTIAL(***且有序);3)EPHEMERAL(臨時(shí));4)EPHEMERAL_SEQUENTIAL(臨時(shí)且有序)。
- internalLockLoop：阻塞等待直到獲得鎖。
看下internalLockLoop是怎么判斷鎖以及阻塞等待的，這里刪除了一些無(wú)關(guān)代碼，只保留主流程：
- //自旋直至獲得鎖
- while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
- {
- //獲取所有的子節(jié)點(diǎn)列表，并且按序號(hào)從小到大排序
- List
children = getSortedChildren(); - //根據(jù)序號(hào)判斷當(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
- {
- //否則獲取前一個(gè)子節(jié)點(diǎn)
- String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
- //這里使用對(duì)象監(jiān)視器做線程同步，當(dāng)獲取不到鎖時(shí)監(jiān)聽(tīng)前一個(gè)子節(jié)點(diǎn)刪除消息并且進(jìn)行wait()，當(dāng)前一個(gè)子節(jié)點(diǎn)刪除（也就是鎖釋放）時(shí)，回調(diào)會(huì)通過(guò)notifyAll喚醒此線程，此線程繼續(xù)自旋判斷是否獲得鎖
- synchronized(this)
- {
- try
- {
- //這里使用getData()接口而不是checkExists()是因?yàn)?，如果前一個(gè)子節(jié)點(diǎn)已經(jīng)被刪除了那么會(huì)拋出異常而且不會(huì)設(shè)置事件監(jiān)聽(tīng)器，而checkExists雖然也可以獲取到節(jié)點(diǎn)是否存在的信息但是同時(shí)設(shè)置了監(jiān)聽(tīng)器，這個(gè)監(jiān)聽(tīng)器其實(shí)永遠(yuǎn)不會(huì)觸發(fā)，對(duì)于zookeeper來(lái)說(shuō)屬于資源泄露
- client.getData().usingWatcher(watcher).forPath(previousSequencePath);
- //如果設(shè)置了阻塞等待的時(shí)間
- if ( millisToWait != null )
- {
- millisToWait -= (System.currentTimeMillis() - startMillis);
- startMillis = System.currentTimeMillis();
- if ( millisToWait <= 0 )
- {
- doDelete = true; // 等待時(shí)間到達(dá)，刪除對(duì)應(yīng)的子節(jié)點(diǎn)
- break;
- }
- //等待相應(yīng)的時(shí)間
- wait(millisToWait);
- }
- else
- {
- //永遠(yuǎn)等待
- wait();
- }
- }
- catch ( KeeperException.NoNodeException e )
- {
- //上面使用getData來(lái)設(shè)置監(jiān)聽(tīng)器時(shí)，如果前一個(gè)子節(jié)點(diǎn)已經(jīng)被刪除那么會(huì)拋出NoNodeException，只需要自旋一次即可，無(wú)需額外處理
- }
- }
- }
- }
具體邏輯見(jiàn)注釋?zhuān)辉儋樖?。代碼中設(shè)置的事件監(jiān)聽(tīng)器，在事件發(fā)生回調(diào)時(shí)只是簡(jiǎn)單的notifyAll喚醒當(dāng)前線程以重新自旋判斷，比較簡(jiǎn)單不再展開(kāi)。
以上。
名稱(chēng)欄目：基于Zookeeper的分布式鎖
當(dāng)前URL：http://www.dlmjj.cn/article/dhhegje.html
