歡迎光臨
每天分享高質量文章

10分鐘看懂!基於Zookeeper的分佈式鎖

精品專欄

 

作者:JVAV曉逸

出處:https://blog.csdn.net/qiangcuo6087/article/details/79067136


實現分佈式鎖目前有三種流行方案,分別為基於資料庫、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:

  1. 客戶端連接zookeeper,併在/lock下創建臨時的且有序的子節點,第一個客戶端對應的子節點為/lock/lock-0000000000,第二個為/lock/lock-0000000001,以此類推。
  2. 客戶端獲取/lock下的子節點串列,判斷自己創建的子節點是否為當前子節點串列中序號最小的子節點,如果是則認為獲得鎖,否則監聽/lock的子節點變更訊息,獲得子節點變更通知後重覆此步驟直至獲得鎖;
  3. 執行業務代碼;
  4. 完成業務流程後,刪除對應的子節點釋放鎖。

步驟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的子節點刪除訊息。

所以調整後的分佈式鎖演算法流程如下:

  • 客戶端連接zookeeper,併在/lock下創建臨時的且有序的子節點,第一個客戶端對應的子節點為/lock/lock-0000000000,第二個為/lock/lock-0000000001,以此類推;
  • 客戶端獲取/lock下的子節點串列,判斷自己創建的子節點是否為當前子節點串列中序號最小的子節點,如果是則認為獲得鎖,否則監聽剛好在自己之前一位的子節點刪除訊息,獲得子節點變更通知後重覆此步驟直至獲得鎖;
  • 執行業務代碼;
  • 完成業務流程後,刪除對應的子節點釋放鎖。

Curator的原始碼分析

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

我們只需要引入下麵這個包(基於maven):

  1.  org.apache.curator
  •  curator-recipes
  •  4.0.0

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

  1. public static void main(String[] args) throws Exception {
  2.  
  3.  //創建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.  //創建分佈式鎖, 鎖空間的根節點路徑為/curator/lock
  12.  
  13.  InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
  14.  
  15.  mutex.acquire();
  16.  
  17.  //獲得了鎖, 進行業務流程
  18.  
  19.  System.out.println("Enter mutex");
  20.  
  21.  //完成業務流程, 釋放鎖
  22.  
  23.  mutex.release();
  24.  
  25.  //關閉客戶端
  26.  
  27.  client.close();
  28.  
  29. }

可以看到關鍵的核心操作就只有mutex.acquire()和mutex.release(),簡直太方便了!

下麵來分析下獲取鎖的原始碼實現。acquire的方法如下:

  1. /*
  2.  * 獲取鎖,當鎖被占用時會阻塞等待,這個操作支持同執行緒的可重入(也就是重覆獲取鎖),acquire的次數需要與release的次數相同。
  3.  * @throws Exception ZK errors, connection interruptions
  4.  */
  5.  
  6. @Override
  7. public void acquire() throws Exception{
  8.  if ( !internalLock(-1, null) ){
  9.      throw new IOException("Lost connection while trying to acquire lock: " + basePath);
  10.    }
  11. }

這裡有個地方需要註意,當與zookeeper通信存在異常時,acquire會直接丟擲異常,需要使用者自身做重試策略。代碼中呼叫了internalLock(-1, null),引數表明在鎖被占用時永久阻塞等待。internalLock的代碼如下

  1. private boolean internalLock(long time, TimeUnit unit) throws Exception {
  2.  
  3.    //這裡處理同執行緒的可重入性,如果已經獲得鎖,那麼只是在對應的資料結構中增加acquire的次數統計,直接傳回成功
  4.    Thread currentThread = Thread.currentThread();
  5.    LockData lockData = threadData.get(currentThread);
  6.    if (lockData != null) {
  7.        lockData.lockCount.incrementAndGet();
  8.        return true;
  9.    }
  10.  
  11.    //這裡才真正去zookeeper中獲取鎖
  12.  
  13.    String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
  14.  
  15.    if (lockPath != null) {
  16.        //獲得鎖之後,記錄當前的執行緒獲得鎖的信息,在重入時只需在LockData中增加次數統計即可
  17.        LockData newLockData = new LockData(currentThread, lockPath);
  18.        threadData.put(currentThread, newLockData);
  19.        return true;
  20.  
  21.    }
  22.  
  23.    //在阻塞傳回時仍然獲取不到鎖,這裡背景關係的處理隱含的意思為zookeeper通信異常
  24.    return false;
  25. }

代碼中增加了具體註釋,不做展開。看下zookeeper獲取鎖的具體實現:

  1. String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception {
  2.  
  3.     //引數初始化,此處省略
  4.     //...
  5.     //自旋獲取鎖
  6.     while (!isDone) {
  7.         isDone = true;
  8.         try {
  9.         //在鎖空間下創建臨時且有序的子節點
  10.             ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
  11.             //判斷是否獲得鎖(子節點序號最小),獲得鎖則直接傳回,否則阻塞等待前一個子節點刪除通知
  12.             hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
  13.         } catch (KeeperException.NoNodeException e) {
  14.  
  15.             //對於NoNodeException,代碼中確保了只有發生session過期才會在這裡丟擲NoNodeException,因此這裡根據重試策略進行重試
  16.             if (client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {
  17.                 isDone = false;
  18.             } else {
  19.                 throw e;
  20.             }
  21.  
  22.         }
  23.  
  24.     }
  25.  
  26.     //如果獲得鎖則傳回該子節點的路徑
  27.     if (hasTheLock) {
  28.         return ourPath;
  29.     }
  30.  
  31.     return null;
  32.  
  33. }

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

  • driver.createsTheLock:創建臨時且有序的子節點,裡面實現比較簡單不做展開,主要關註幾種節點的樣式:1)PERSISTENT(永久);2)PERSISTENTSEQUENTIAL(永久且有序);3)EPHEMERAL(臨時);4)EPHEMERALSEQUENTIAL(臨時且有序)。
  • internalLockLoop:阻塞等待直到獲得鎖。

看下internalLockLoop是怎麼判斷鎖以及阻塞等待的,這裡刪除了一些無關代碼,只保留主流程:

  1. //自旋直至獲得鎖
  2. while((client.getState()==CuratorFrameworkState.STARTED)&&!haveTheLock ){
  3.     //獲取所有的子節點串列,並且按序號從小到大排序
  4.     List<String> children = getSortedChildren();
  5.  
  6.     //根據序號判斷當前子節點是否為最小子節點
  7.     String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
  8.     PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
  9.  
  10.     if (predicateResults.getsTheLock()) {
  11.         //如果為最小子節點則認為獲得鎖
  12.         haveTheLock = true;
  13.     } else {
  14.         //否則獲取前一個子節點
  15.         String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
  16.  
  17.         //這裡使用物件監視器做執行緒同步,當獲取不到鎖時監聽前一個子節點刪除訊息並且進行wait(),當前一個子節點刪除(也就是鎖釋放)時,
  18.         // 回呼會通過notifyAll喚醒此執行緒,此執行緒繼續自旋判斷是否獲得鎖
  19.         synchronized (this) {
  20.             try {
  21.                 //這裡使用getData()接口而不是checkExists()是因為,如果前一個子節點已經被刪除了那麼會丟擲異常而且不會設置事件監聽器,
  22.                 // 而checkExists雖然也可以獲取到節點是否存在的信息但是同時設置了監聽器,這個監聽器其實永遠不會觸發,對於zookeeper來說屬於資源泄露
  23.                 client.getData().usingWatcher(watcher).forPath(previousSequencePath);
  24.  
  25.                 //如果設置了阻塞等待的時間
  26.                 if (millisToWait != null) {
  27.                     millisToWait -= (System.currentTimeMillis() - startMillis);
  28.                     startMillis = System.currentTimeMillis();
  29.                     if (millisToWait <= 0) {
  30.                         doDelete = true; // 等待時間到達,刪除對應的子節點
  31.                         break;
  32.                     }
  33.                     //等待相應的時間
  34.                     wait(millisToWait);
  35.                 } else {
  36.                     //永遠等待
  37.                     wait();
  38.  
  39.                 }
  40.  
  41.             } catch (KeeperException.NoNodeException e) {
  42.                 //上面使用getData來設置監聽器時,如果前一個子節點已經被刪除那麼會丟擲NoNodeException,只需要自旋一次即可,無需額外處理
  43.             }
  44.         }
  45.     }
  46. }

具體邏輯見註釋,不再贅述。代碼中設置的事件監聽器,在事件發生回呼時只是簡單的notifyAll喚醒當前執行緒以重新自旋判斷,比較簡單不再展開。

Core Java 併發:理解併發概念

分分鐘解決 MySQL 查詢速度慢與性能差

Java 必須掌握的 20+ 種 Spring 常用註解

為什麼阿裡巴巴禁止開發人員使用isSuccess作為變數名

8種常被忽視的SQL錯誤用法

可能是最全面的G1學習筆記

END

>>>>>> 加群交流技術 <<<<<<

    閱讀原文

    赞(0)

    分享創造快樂