|
實(shí)現(xiàn)分布式鎖目前有三種流行方案，分別為基于數(shù)據(jù)庫、Redis、Zookeeper的方案，其中前兩種方案網(wǎng)絡(luò)上有很多資料可以參考，本文不做展開。我們來看下使用Zookeeper如何實(shí)現(xiàn)分布式鎖。 什么是Zookeeper？ Zookeeper（業(yè)界簡稱zk）是一種提供配置管理、分布式協(xié)同以及命名的中心化服務(wù)，這些提供的功能都是分布式系統(tǒng)中非常底層且必不可少的基本功能，但是如果自己實(shí)現(xiàn)這些功能而且要達(dá)到高吞吐、低延遲同時(shí)還要保持一致性和可用性，實(shí)際上非常困難。因此zookeeper提供了這些功能，開發(fā)者在zookeeper之上構(gòu)建自己的各種分布式系統(tǒng)。 雖然zookeeper的實(shí)現(xiàn)比較復(fù)雜，但是它提供的模型抽象卻是非常簡單的。Zookeeper提供一個(gè)多層級的節(jié)點(diǎn)命名空間（節(jié)點(diǎn)稱為znode），每個(gè)節(jié)點(diǎn)都用一個(gè)以斜杠（/）分隔的路徑表示，而且每個(gè)節(jié)點(diǎn)都有父節(jié)點(diǎn)（根節(jié)點(diǎn)除外），非常類似于文件系統(tǒng)。例如，/foo/doo這個(gè)表示一個(gè)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ù)了這個(gè)樹狀的目錄結(jié)構(gòu)，這種特性使得Zookeeper不能用于存放大量的數(shù)據(jù)，每個(gè)節(jié)點(diǎn)的存放數(shù)據(jù)上限為1M。 而為了保證高可用，zookeeper需要以集群形態(tài)來部署，這樣只要集群中大部分機(jī)器是可用的（能夠容忍一定的機(jī)器故障），那么zookeeper本身仍然是可用的?？蛻舳嗽谑褂脄ookeeper時(shí)，需要知道集群機(jī)器列表，通過與集群中的某一臺機(jī)器建立TCP連接來使用服務(wù)，客戶端使用這個(gè)TCP鏈接來發(fā)送請求、獲取結(jié)果、獲取監(jiān)聽事件以及發(fā)送心跳包。如果這個(gè)連接異常斷開了，客戶端可以連接到另外的機(jī)器上。 架構(gòu)簡圖如下所示： 
zookeeper 客戶端的讀請求可以被集群中的任意一臺機(jī)器處理，如果讀請求在節(jié)點(diǎn)上注冊了監(jiān)聽器，這個(gè)監(jiān)聽器也是由所連接的zookeeper機(jī)器來處理。對于寫請求，這些請求會(huì)同時(shí)發(fā)給其他zookeeper機(jī)器并且達(dá)成一致后，請求才會(huì)返回成功。因此，隨著zookeeper的集群機(jī)器增多，讀請求的吞吐會(huì)提高但是寫請求的吞吐會(huì)下降。 有序性是zookeeper中非常重要的一個(gè)特性，所有的更新都是全局有序的，每個(gè)更新都有一個(gè)唯一的時(shí)間戳，這個(gè)時(shí)間戳稱為zxid（Zookeeper Transaction Id）。而讀請求只會(huì)相對于更新有序，也就是讀請求的返回結(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ù)序號，也就是說如果是第一個(gè)創(chuàng)建的子節(jié)點(diǎn)，那么生成的子節(jié)點(diǎn)為/lock/node-0000000000，下一個(gè)節(jié)點(diǎn)則為/lock/node-0000000001，依次類推。 臨時(shí)節(jié)點(diǎn)：客戶端可以建立一個(gè)臨時(shí)節(jié)點(diǎn)，在會(huì)話結(jié)束或者會(huì)話超時(shí)后，zookeeper會(huì)自動(dòng)刪除該節(jié)點(diǎn)。 事件監(jiān)聽：在讀取數(shù)據(jù)時(shí)，我們可以同時(shí)對節(jié)點(diǎn)設(shè)置事件監(jiān)聽，當(dāng)節(jié)點(diǎn)數(shù)據(jù)或結(jié)構(gòu)變化時(shí)，zookeeper會(huì)通知客戶端。當(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)建臨時(shí)的且有序的子節(jié)點(diǎn)，第一個(gè)客戶端對應(yīng)的子節(jié)點(diǎn)為/lock/lock-0000000000，第二個(gè)為/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)建的臨時(shí)節(jié)點(diǎn)能夠保證在故障的情況下鎖也能被釋放，考慮這么個(gè)場景：假如客戶端a當(dāng)前創(chuàng)建的子節(jié)點(diǎn)為序號最小的節(jié)點(diǎn)，獲得鎖之后客戶端所在機(jī)器宕機(jī)了，客戶端沒有主動(dòng)刪除子節(jié)點(diǎn)；如果創(chuàng)建的是永久的節(jié)點(diǎn)，那么這個(gè)鎖永遠(yuǎn)不會(huì)釋放，導(dǎo)致死鎖；由于創(chuàng)建的是臨時(shí)節(jié)點(diǎn)，客戶端宕機(jī)后，過了一定時(shí)間zookeeper沒有收到客戶端的心跳包判斷會(huì)話失效，將臨時(shí)節(jié)點(diǎn)刪除從而釋放鎖。 另外細(xì)心的朋友可能會(huì)想到，在步驟2中獲取子節(jié)點(diǎn)列表與設(shè)置監(jiān)聽這兩步操作的原子性問題，考慮這么個(gè)場景：客戶端a對應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000000，客戶端b對應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000001，客戶端b獲取子節(jié)點(diǎn)列表時(shí)發(fā)現(xiàn)自己不是序號最小的，但是在設(shè)置監(jiān)聽器前客戶端a完成業(yè)務(wù)流程刪除了子節(jié)點(diǎn)/lock/lock-0000000000，客戶端b設(shè)置的監(jiān)聽器豈不是丟失了這個(gè)事件從而導(dǎo)致永遠(yuǎn)等待了？這個(gè)問題不存在的。因?yàn)閦ookeeper提供的API中設(shè)置監(jiān)聽器的操作與讀操作是原子執(zhí)行的，也就是說在讀子節(jié)點(diǎn)列表時(shí)同時(shí)設(shè)置監(jiān)聽器，保證不會(huì)丟失事件。 最后，對于這個(gè)算法有個(gè)極大的優(yōu)化點(diǎn)：假如當(dāng)前有1000個(gè)節(jié)點(diǎn)在等待鎖，如果獲得鎖的客戶端釋放鎖時(shí)，這1000個(gè)客戶端都會(huì)被喚醒，這種情況稱為“羊群效應(yīng)”；在這種羊群效應(yīng)中，zookeeper需要通知1000個(gè)客戶端，這會(huì)阻塞其他的操作，最好的情況應(yīng)該只喚醒新的最小節(jié)點(diǎn)對應(yīng)的客戶端。應(yīng)該怎么做呢？在設(shè)置事件監(jiān)聽時(shí)，每個(gè)客戶端應(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)建臨時(shí)的且有序的子節(jié)點(diǎn)，第一個(gè)客戶端對應(yīng)的子節(jié)點(diǎn)為/lock/lock-0000000000，第二個(gè)為/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)一個(gè)分布式鎖還是比較麻煩的…我們可以直接使用curator這個(gè)開源項(xiàng)目提供的zookeeper分布式鎖實(shí)現(xiàn)。 我們只需要引入下面這個(gè)包（基于maven）： <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.0.0</version> </dependency>
然后就可以用啦！代碼如下： public static void main(String[] args) throws Exception { //創(chuàng)建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(); //創(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)閉客戶端 client.close(); }
可以看到關(guān)鍵的核心操作就只有mutex.acquire()和mutex.release()，簡直太方便了！ 下面來分析下獲取鎖的源碼實(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)獲得鎖，那么只是在對應(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; }
代碼中增加了具體注釋，不做展開。看下zookeeper獲取鎖的具體實(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)序號最?。?，獲得鎖則直接返回，否則阻塞等待前一個(gè)子節(jié)點(diǎn)刪除通知 hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath); } catch ( KeeperException.NoNodeException e ) { //對于NoNodeException，代碼中確保了只有發(fā)生session過期才會(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)比較簡單不做展開，主要關(guān)注幾種節(jié)點(diǎn)的模式：1）PERSISTENT（永久）；2）PERSISTENT_SEQUENTIAL（永久且有序）；3）EPHEMERAL（臨時(shí)）；4）EPHEMERAL_SEQUENTIAL（臨時(shí)且有序）。 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 { //否則獲取前一個(gè)子節(jié)點(diǎn) String previousSequencePath = basePath + '/' + predicateResults.getPathToWatch(); //這里使用對象監(jiān)視器做線程同步，當(dāng)獲取不到鎖時(shí)監(jiān)聽前一個(gè)子節(jié)點(diǎn)刪除消息并且進(jìn)行wait()，當(dāng)前一個(gè)子節(jié)點(diǎn)刪除（也就是鎖釋放）時(shí)，回調(diào)會(huì)通過notifyAll喚醒此線程，此線程繼續(xù)自旋判斷是否獲得鎖 synchronized(this) { try { //這里使用getData()接口而不是checkExists()是因?yàn)椋绻耙粋€(gè)子節(jié)點(diǎn)已經(jīng)被刪除了那么會(huì)拋出異常而且不會(huì)設(shè)置事件監(jiān)聽器，而checkExists雖然也可以獲取到節(jié)點(diǎn)是否存在的信息但是同時(shí)設(shè)置了監(jiān)聽器，這個(gè)監(jiān)聽器其實(shí)永遠(yuǎn)不會(huì)觸發(fā)，對于zookeeper來說屬于資源泄露 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á)，刪除對應(yīng)的子節(jié)點(diǎn) break; } //等待相應(yīng)的時(shí)間 wait(millisToWait); } else { //永遠(yuǎn)等待 wait(); } } catch ( KeeperException.NoNodeException e ) { //上面使用getData來設(shè)置監(jiān)聽器時(shí)，如果前一個(gè)子節(jié)點(diǎn)已經(jīng)被刪除那么會(huì)拋出NoNodeException，只需要自旋一次即可，無需額外處理 } } } }
具體邏輯見注釋，不再贅述。代碼中設(shè)置的事件監(jiān)聽器，在事件發(fā)生回調(diào)時(shí)只是簡單的notifyAll喚醒當(dāng)前線程以重新自旋判斷，比較簡單不再展開。 Java學(xué)習(xí)資料獲?。◤?fù)制下段連接至瀏覽器即可）
data:text/html;charset=UTF-8;base64,5oGt5Zac5L2g77yM5p625p6E5biI5a2m5Lmg576k5Y+35pivNjg2NTc5MDE0Cg==
|
