溫馨提示×

溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊×
其他方式登錄
點(diǎn)擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》

基于Zookeeper的分布式鎖該如何理解

發(fā)布時(shí)間：2021-12-24 16:40:43 來源：億速云 閱讀：144 作者：柒染 欄目：互聯(lián)網(wǎng)科技

今天就跟大家聊聊有關(guān)基于Zookeeper的分布式鎖該如何理解，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

實(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è)多層級(jí)的節(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ī)器列表，通過與集群中的某一臺(tái)機(jī)器建立TCP連接來使用服務(wù)，客戶端使用這個(gè)TCP鏈接來發(fā)送請求、獲取結(jié)果、獲取監(jiān)聽事件以及發(fā)送心跳包。如果這個(gè)連接異常斷開了，客戶端可以連接到另外的機(jī)器上。

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

基于Zookeeper的分布式鎖該如何理解

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

有序性是zookeeper中非常重要的一個(gè)特性，所有的更新都是全局有序的，每個(gè)更新都有一個(gè)唯一的時(shí)間戳，這個(gè)時(shí)間戳稱為zxid（Zookeeper Transaction Id）。而讀請求只會(huì)相對(duì)于更新有序，也就是讀請求的返回結(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)，也就是說如果是第一個(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í)對(duì)節(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的分布式鎖該如何理解

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

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

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

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

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

步驟1中創(chuàng)建的臨時(shí)節(jié)點(diǎn)能夠保證在故障的情況下鎖也能被釋放，考慮這么個(gè)場景：假如客戶端a當(dāng)前創(chuàng)建的子節(jié)點(diǎn)為序號(hào)最小的節(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對(duì)應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000000，客戶端b對(duì)應(yīng)子節(jié)點(diǎn)為/lock/lock-0000000001，客戶端b獲取子節(jié)點(diǎn)列表時(shí)發(fā)現(xiàn)自己不是序號(hào)最小的，但是在設(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ì)丟失事件。

最后，對(duì)于這個(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)對(duì)應(yīng)的客戶端。應(yīng)該怎么做呢？在設(shè)置事件監(jiān)聽時(shí)，每個(gè)客戶端應(yīng)該對(duì)剛好在它之前的子節(jié)點(diǎn)設(shè)置事件監(jiān)聽，例如子節(jié)點(diǎn)列表為/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序號(hào)為1的客戶端監(jiān)聽序號(hào)為0的子節(jié)點(diǎn)刪除消息，序號(hào)為2的監(jiān)聽序號(hào)為1的子節(jié)點(diǎn)刪除消息。

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

  • 客戶端連接zookeeper，并在/lock下創(chuàng)建臨時(shí)的且有序的子節(jié)點(diǎn)，第一個(gè)客戶端對(duì)應(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)列表中序號(hào)最小的子節(jié)點(diǎn)，如果是則認(rèn)為獲得鎖，否則監(jiān)聽剛好在自己之前一位的子節(jié)點(diǎn)刪除消息，獲得子節(jié)點(diǎn)變更通知后重復(fù)此步驟直至獲得鎖；

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

  • 完成業(yè)務(wù)流程后，刪除對(duì)應(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>

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

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

  3.   //創(chuàng)建zookeeper的客戶端

  5.   RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

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

  9.   client.start();

  11.   //創(chuàng)建分布式鎖, 鎖空間的根節(jié)點(diǎn)路徑為/curator/lock

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

  15.   mutex.acquire();

  17.   //獲得了鎖, 進(jìn)行業(yè)務(wù)流程

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

  21.   //完成業(yè)務(wù)流程, 釋放鎖

  23.   mutex.release();

  25.   //關(guān)閉客戶端

  27.   client.close();

  29. }

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

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

  1. /*
  2.   * 獲取鎖，當(dāng)鎖被占用時(shí)會(huì)阻塞等待，這個(gè)操作支持同線程的可重入（也就是重復(fù)獲取鎖），acquire的次數(shù)需要與release的次數(shù)相同。
  3.   * @throws Exception ZK errors, connection interruptions
  4.   */

  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. }

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

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

  3.     //這里處理同線程的可重入性，如果已經(jīng)獲得鎖，那么只是在對(duì)應(yīng)的數(shù)據(jù)結(jié)構(gòu)中增加acquire的次數(shù)統(tǒng)計(jì)，直接返回成功
  4.     Thread currentThread = Thread.currentThread();
  5.     LockData lockData = threadData.get(currentThread);
  6.     if (lockData != null) {
  7.         lockData.lockCount.incrementAndGet();
  8.         return true;
  9.     }

  11.     //這里才真正去zookeeper中獲取鎖

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

  15.     if (lockPath != null) {
  16.         //獲得鎖之后，記錄當(dāng)前的線程獲得鎖的信息，在重入時(shí)只需在LockData中增加次數(shù)統(tǒng)計(jì)即可
  17.         LockData newLockData = new LockData(currentThread, lockPath);
  18.         threadData.put(currentThread, newLockData);
  19.         return true;

  21.     }

  23.     //在阻塞返回時(shí)仍然獲取不到鎖，這里上下文的處理隱含的意思為zookeeper通信異常
  24.     return false;
  25. }

代碼中增加了具體注釋，不做展開?？聪聑ookeeper獲取鎖的具體實(shí)現(xiàn)：

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

  3.      //參數(shù)初始化，此處省略
  4.      //...
  5.      //自旋獲取鎖
  6.      while (!isDone) {
  7.          isDone = true;
  8.          try {
  9.          //在鎖空間下創(chuàng)建臨時(shí)且有序的子節(jié)點(diǎn)
  10.              ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
  11.              //判斷是否獲得鎖（子節(jié)點(diǎn)序號(hào)最?。@得鎖則直接返回，否則阻塞等待前一個(gè)子節(jié)點(diǎn)刪除通知
  12.              hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
  13.          } catch (KeeperException.NoNodeException e) {

  15.              //對(duì)于NoNodeException，代碼中確保了只有發(fā)生session過期才會(huì)在這里拋出NoNodeException，因此這里根據(jù)重試策略進(jìn)行重試
  16.              if (client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {
  17.                  isDone = false;
  18.              } else {
  19.                  throw e;
  20.              }

  22.          }

  24.      }

  26.      //如果獲得鎖則返回該子節(jié)點(diǎn)的路徑
  27.      if (hasTheLock) {
  28.          return ourPath;
  29.      }

  31.      return null;

  33.  }

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

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

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

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

  1. //自旋直至獲得鎖
  2.  while((client.getState()==CuratorFrameworkState.STARTED)&&!haveTheLock ){
  3.      //獲取所有的子節(jié)點(diǎn)列表，并且按序號(hào)從小到大排序
  4.      List<String> children = getSortedChildren();

  6.      //根據(jù)序號(hào)判斷當(dāng)前子節(jié)點(diǎn)是否為最小子節(jié)點(diǎn)
  7.      String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
  8.      PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);

  10.      if (predicateResults.getsTheLock()) {
  11.          //如果為最小子節(jié)點(diǎn)則認(rèn)為獲得鎖
  12.          haveTheLock = true;
  13.      } else {
  14.          //否則獲取前一個(gè)子節(jié)點(diǎn)
  15.          String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();

  17.          //這里使用對(duì)象監(jiān)視器做線程同步，當(dāng)獲取不到鎖時(shí)監(jiān)聽前一個(gè)子節(jié)點(diǎn)刪除消息并且進(jìn)行wait()，當(dāng)前一個(gè)子節(jié)點(diǎn)刪除（也就是鎖釋放）時(shí)，
  18.          // 回調(diào)會(huì)通過notifyAll喚醒此線程，此線程繼續(xù)自旋判斷是否獲得鎖
  19.          synchronized (this) {
  20.              try {
  21.                  //這里使用getData()接口而不是checkExists()是因?yàn)?，如果前一個(gè)子節(jié)點(diǎn)已經(jīng)被刪除了那么會(huì)拋出異常而且不會(huì)設(shè)置事件監(jiān)聽器，
  22.                  // 而checkExists雖然也可以獲取到節(jié)點(diǎn)是否存在的信息但是同時(shí)設(shè)置了監(jiān)聽器，這個(gè)監(jiān)聽器其實(shí)永遠(yuǎn)不會(huì)觸發(fā)，對(duì)于zookeeper來說屬于資源泄露
  23.                  client.getData().usingWatcher(watcher).forPath(previousSequencePath);

  25.                  //如果設(shè)置了阻塞等待的時(shí)間
  26.                  if (millisToWait != null) {
  27.                      millisToWait -= (System.currentTimeMillis() - startMillis);
  28.                      startMillis = System.currentTimeMillis();
  29.                      if (millisToWait <= 0) {
  30.                          doDelete = true; // 等待時(shí)間到達(dá)，刪除對(duì)應(yīng)的子節(jié)點(diǎn)
  31.                          break;
  32.                      }
  33.                      //等待相應(yīng)的時(shí)間
  34.                      wait(millisToWait);
  35.                  } else {
  36.                      //永遠(yuǎn)等待
  37.                      wait();

  39.                  }

  41.              } catch (KeeperException.NoNodeException e) {
  42.                  //上面使用getData來設(shè)置監(jiān)聽器時(shí)，如果前一個(gè)子節(jié)點(diǎn)已經(jīng)被刪除那么會(huì)拋出NoNodeException，只需要自旋一次即可，無需額外處理
  43.              }
  44.          }
  45.      }
  46.  }

代碼中設(shè)置的事件監(jiān)聽器，在事件發(fā)生回調(diào)時(shí)只是簡單的notifyAll喚醒當(dāng)前線程以重新自旋判斷，比較簡單不再展開。

看完上述內(nèi)容，你們對(duì)基于Zookeeper的分布式鎖該如何理解有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。

向AI問一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手