Curatorのソースコードから群集効果まで

1. はじめに

Curator は、Zookeeper を操作し、分散ロックや選挙などの高度な機能をカプセル化する、Java で記述されたクライアント ツールです。

今日は主に分散ロックの主な原理を分析します。分散ロックの紹介やその他の実装については、興味のある学生は以下の記事をお読みください。

私は数万語を費やして、Redis で分散ロックを実装する困難な道を、単一マシンからマスタースレーブ、複数インスタンスまで歩き回りましたが、多くの問題が発生することがわかりました_Yangyang のブログ - CSDN ブログ

Redisson の再入とロック更新のソース コード分析_Yang Yang のブログ - CSDN ブログ

Curator を使用して分散ロックを取得する場合、Curator は指定されたパスの下に順序付けられた一時ノードを作成します。ノードが最小の場合、ロックが正常に取得されたことを意味します。

次に、準備として、一時ノードが作成されるかどうかを確認します。

2. 準備

まず、Zookeeper クラスターを構築する必要がありますが、もちろん単一のマシンを使用することもできます。

この記事では、「インタビューアー：Zookeeper 選挙の図を描いてもらえますか？」というタイトルで、docker-compose を使って素早く zk クラスターを構築する方法が紹介されています。

pom に依存関係を導入します。

 <依存関係>
         <グループ ID>org.apache.curator</グループ ID>
         <artifactId>キュレーターレシピ</artifactId>
         <バージョン>2.12.0</バージョン>
     </依存関係>

Curator クライアントの構成項目:

 /**
 * @著者 qcy
 *@作成2022/01/01 22:59:34
 */
 @構成
パブリッククラス CuratorFrameworkConfig { 
 
    //zk ノードアドレス
    プライベート静的最終文字列 CONNECT_STRING = "localhost:2181,localhost:2182,localhost:2183" ;
    //接続タイムアウト（単位：ミリ秒）
    プライベート静的最終int CONNECTION_TIME_OUT_MS = 10 * 1000;
    //セッションタイムアウト（単位：ミリ秒）
    プライベート静的最終int SESSION_TIME_OUT_MS = 30 * 1000;
    //再試行の初期待機時間（単位：ミリ秒）
    プライベート静的最終int BASE_SLEEP_TIME_MS = 2 * 1000;
    //再試行の最大回数
    プライベート静的最終int MAX_RETRIES = 3; 
 
    @ビーン
パブリックCuratorFramework getCuratorFramework() {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(接続文字列)
                .connectionTimeoutMs(接続タイムアウトMS)
                .sessionTimeoutMs(セッションタイムアウトMS)
                .retryPolicy(新しいExponentialBackoffRetry(BASE_SLEEP_TIME_MS、MAX_RETRIES))
                。建てる（）;
        curatorFramework を開始します。
 curatorFrameworkを返します。
    } 
     
 }

SESSION_TIME_OUT_MS パラメータは、クライアントがロックを取得した後に突然クラッシュした場合でも、zk がその時間内に現在のクライアントによって作成された一時的な順序付きノードを削除できるようにします。

テストコードは次のとおりです。

 // 一時的なノードパス、qcy はブロガーの名前の略称です
   プライベート静的最終文字列 LOCK_PATH = "/lockqcy" ; 
 
   @リソース
   キュレーターフレームワーク キュレーターフレームワーク; 
 
パブリックvoid testCurator() 例外をスローします {
       InterProcessMutex interProcessMutex = 新しい InterProcessMutex(curatorFramework, LOCK_PATH);
       インタープロセスミューテックスを取得します。 
 
       試す {
           //シミュレーション業務時間
           スレッドをスリープ状態にします(30 * 1000);
       } キャッチ (例外 e) {
           e.printStackTrace();
       ついに
           プロセスミューテックスを解放します。
       }
   }

インターフェースを使用してこのメ​​ソッドを呼び出すときは、Thread.sleep にブレークポイントを設定し、zk コンテナーに入って作成されたノードを観察します。

docker exec -it zk container name/bin/bash を使用してコンテナに対話モードで入り、./bin/zkCli.sh を使用して zk サーバーに接続します。

次にls pathを使用してノードを表示します

これら 3 つのノードは永続ノードです。 get path を使用して、ノードのデータ構造情報を表示できます。

ノードの ephemeralOwner 値が 0 の場合、つまりノードの一時所有者のセッション ID が 0 の場合、そのノードは永続ノードであることを意味します。

ブレークポイント Thread.sleep に到達したとき、lockqcy の下に一時ノードが作成されたことがわかりました。

この時点で準備は完了です。次に、interProcessMutex.acquireとreleaseのプロセスを分析してみましょう。

3. ソースコード分析

Curatorは、次のような複数のタイプのロックをサポートしています。

• InterProcessMutex、再入可能ロック、排他ロック
• InterProcessReadWriteLock、読み取り書き込みロック
• InterProcessSemaphoreMutex、非再入可能排他ロック

今日は主に InterProcessMutex のロックとロック解除のプロセスを分析します。まずロックのプロセスを見てみましょう

ロック

パブリックvoid acquire() は例外をスローします {
      内部ロックが-1の場合、 nullの場合
          throw new IOException( "ロックを取得中に接続が失われました: " + basePath);
      }
  }

これはブロッキング ロックの取得です。ロックを取得できない場合は、ブロックが継続されます。したがって、internalLock メソッドの場合、タイムアウトは -1 に設定され、時間単位は null に設定されます。

 private boolean internalLock(long time , TimeUnit unit) は例外をスローします {
       スレッド currentThread = Thread.currentThread();
       //スレッドのLockData情報がマップから取得できるかどうかによって、スレッドがすでにロックを保持しているかどうかを判断します。
       ロックデータ lockData = threadData.get(currentThread);
       ロックデータ != null の場合{
           //再入可能、ロック成功に直接戻る
           ロックデータ.ロックカウント.増加および取得();
戻る 真実;
       }
       //ロック
       文字列 lockPath = internals.attemptLock( time , unit, getLockNodeBytes());
       lockPath != null の場合{
           //ロックに成功しました。マップに保存します
           LockData newLockData = 新しい LockData(currentThread、lockPath);
           threadData.put(現在のスレッド、新しいロックデータ);
戻る 真実;
       } 
 
戻る 間違い;
   }

このうち、threadData はマップであり、キーはスレッド オブジェクト、値はスレッドにバインドされたロック データです。

LockDataは、ロックスレッドowningThread、再入カウントlockCount、およびロックパスlockPathを格納します。例:

 /lockqcy/_c_c46513c3-ace0-405f-aa1e-a531ce28fb47-ロック-0000000005

プライベート最終 ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap(); 
 
    プライベート静的クラスLockData {
        最終スレッド所有スレッド;
        最終的な文字列 lockPath;
        最終的なAtomicInteger lockCount = 新しいAtomicInteger(1); 
 
        プライベートLockData(スレッド所有スレッド、文字列lockPath) {
            this.owningThread = 所有しているスレッド;
            ロックパスをロックします。
        }
    }

internals.attemptLockメソッドを入力する

文字列 attemptLock(long time , TimeUnit unit, byte[] lockNodeBytes) は例外をスローします {
      //開始時間
      最終的な長いstartMillis = System.currentTimeMillis();
      //タイムアウト期間をミリ秒に変換します
      最終的な Long millisToWait = (unit != null ) ?ユニット.toMillis(時間) : null ;
      //ノードデータ、ここはnull  
      最終的なbyte[] localLockNodeBytes = (revocable.get() != null ) ?新しいバイト[0]: lockNodeBytes;
      //再試行回数
再試行回数 = 0;
      //パスをロック
      文字列 ourPath = null ;
      //ロックを取得するかどうか
      ブール値hasTheLock = false ;
      //完了しましたか?
      ブール値 isDone = false ; 
 
      while (!isDone) {
          完了 = true ; 
 
          試す {
              // 一時的に順序付けられたノードを作成し、ノードのパスを返します
              // 内部的にクライアントを呼び出します。作成().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
              ourPath = driver.createsTheLock(クライアント、パス、localLockNodeBytes);
              //返されたノードパスに基づいて、ロックが取得されているかどうかを判断します
              ロックがある = internalLockLoop(startMillis、millisToWait、ourPath);
          } キャッチ (KeeperException.NoNodeException e) {
              //セッションが期限切れになると、ドライバーは一時的に順序付けられたノードを見つけられなくなり、NoNodeException をスローすることがあります。
              //ここで再試行
              (client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++、System.currentTimeMillis() - startMillis、RetryLoop.getDefaultRetrySleeper())) {
                  isDone = false ;
              }それ以外{
                  eを投げる;
              }
          }
      }
      //ロックを取得し、呼び出し元がマップに記録するためのノード パスを返します。
      ロックがある場合
ourPathを返します。
      } 
 
戻る ヌル;
  }

次に、先ほど作成した一時的な順序付きノードが internalLockLoop で使用され、ロックが取得されたかどうかが判断されます。

 private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) は例外をスローします {
       //ロックを取得するかどうか
       ブール値 haveTheLock = false ;
       ブール型doDelete = false ;
       試す {
           (revocable.get() != null ) の場合 {
               //今回はここには入りません
               client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
           }
           //ロックの取得を試行し続ける
           ((client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock) の間 {
               //basePath（ここではlockqcy）の下のすべての一時的に順序付けられたノードを返し、サフィックスに従って昇順に並べます
               リスト<String> children = getSortedChildren();
               //現在のスレッドによって作成された一時的な順序付きノードの名前を取得します。ここでは /_c_c46513c3-ace0-405f-aa1e-a531ce28fb47-lock-0000000005 です。
               文字列sequenceNodeName = ourPath.substring ( basePath.length () + 1);
               //ソート後に現在のノードが 1 位であるかどうかを判断します。そもそもロックが取得されていることを意味する
               述語結果 predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
               if (述語結果.getsTheLock()) {
                   //ロックを取得したらループを終了する
                   ロックを有効にする = true ;
               }それ以外{
                   //これはロックが取得されなかったことを意味します
                   //現在のノードより小さいインデックスを持つ前のノードを取得します
                   文字列 previousSequencePath = basePath + "/" + predicateResults.getPathToWatch(); 
 
                   同期された（これ）{
                       試す {
                           //前のノードが存在しない場合は、NoNodeExceptionを直接スローし、catchで処理せず、次のラウンドでロックの取得を続行します
                           // 前のノードが存在する場合は、そのノードの解放イベントをリッスンするリスナーを設定します
                           client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                           （ミリストゥウェイト！= null ）の場合{
                               millisToWait -= (System.currentTimeMillis() - startMillis);
                               startMillis = System.currentTimeMillis();
                               //タイムアウトしたか判断する
                               （ミリ秒待機時間 <= 0）の場合{
                                   //ロックタイムアウトを取得し、作成した一時的な順序付きノードを削除します
                                   削除する = true ;
                                   壊す;
                               }
                               //タイムアウトがない場合は、millisToWait で待機します
                               待機(ミリ秒単位の待機)。
                           }それ以外{
                               // 無期限にブロックして待機し、前のノードが削除された場合にのみウェイクアップ操作がトリガーされます。
                               待って（）;
                           }
                       } キャッチ (KeeperException.NoNodeException e) {
                           //前のノードが存在しない場合は、NoNodeExceptionを直接スローし、catchで処理せず、次のラウンドでロックの取得を続行します
                       }
                   }
               }
           }
       } キャッチ (例外 e) {
           ThreadUtils.checkInterrupted(e);
           削除する = true ;
           eを投げる;
       ついに
           削除する場合
               // 先ほど作成した一時的な順序付きノードを削除します
               パスを削除します。
           }
       }
 haveTheLock を返します。
   }

ロックが取得されたかどうかを判断するためのコアロジックは、getsTheLockにあります。

パブリックPredicateResultsはLockを取得します(CuratorFrameworkクライアント、List<String> children、StringsequenceNodeName、 int maxLeases)例外をスローします{
     // すべての子ノードがソートされた後の現在のノードのインデックス位置を取得します
int ourIndex = children.indexOf(sequenceNodeName);
     // 現在のノードが子ノードにあるかどうかを判定する
     シーケンスノード名、インデックスを検証します。
     //InterProcessMutexの構築メソッドはmaxLeasesを1に初期化します
     //getsTheLock をtrue にするには、ourIndex は 0 である必要があります。つまり、ロックが取得されたことを示すには、現在のノードが basePath の下の最小のノードである必要があります。
     ブール値 getsTheLock = ourIndex < maxLeases;
     //ロックを取得できない場合は、前のノードの名前が返され、そのノードのモニターが設定されます
     文字列 pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases); 
 
新しい PredicateResults(pathToWatch、getsTheLock)を返します。
 } 
 
静的voidのvalidateOurIndex(StringのsequenceNodeName、 intのourIndex)はKeeperExceptionをスローします{
     （ourIndex < 0）の場合{
         //接続が失われると一時ノードが削除される可能性があるため、これは安全対策です
         新しい KeeperException.NoNodeException( "シーケンシャル パスが見つかりません: " + sequenceNodeName);をスローします。
     }
 }

では、internalLockLoop で待機しているスレッドはいつ起動できるのでしょうか?

internalLockLoopメソッドでは、

 client.getData().usingWatcher(watcher).forPath(前のシーケンスパス);

前のノードにリスナーが設定されます。ノードが削除されると、ウォッチャー内のコールバックがトリガーされます。

プライベートファイナルウォッチャーウォッチャー = 新しいウォッチャー() {
        //コールバックメソッド
        @オーバーライド
パブリックvoid プロセス(WatchedEvent イベント) {
            通知からWatcher();
        }
    }; 
 
    プライベート同期voidnotifyFromWatcher() {
        //LockInternalsインスタンスを待機しているすべてのスレッドを起動します
        すべて通知します();
    }

この時点で、ロック プロセスは基本的に分析されており、その概要は次のとおりです。

まず、一時的な順序付きノードを作成します

ノードが basePath 下の最小のノードである場合、ロックが取得されてマップに格納され、次回直接再入力されることを意味します。

ノードが最小のノードでない場合は、前のノードのリスナーを設定して待機します。前のノードが削除されると、通知スレッドに通知されます。

ロック解除

ロック解除のロジックは比較的単純で、リリースメソッドに直接進むだけです

パブリックvoid release() は例外をスローします {
      スレッド currentThread = Thread.currentThread();
      ロックデータ lockData = threadData.get(currentThread);
      ロックデータ== nullの場合
          throw new IllegalMonitorStateException( "ロックを所有していません: " + basePath);
      } 
 
 int newLockCount = lockData.lockCount.decrementAndGet();
      //再突入回数を直接減らす
      (新しいロックカウント > 0) の場合 {
戻る;
      }
      (新しいロックカウント < 0) の場合 {
          throw new IllegalMonitorStateException( "ロックのロック数が負になりました: " + basePath);
      } 
 
      //ここでは再入回数が0であることを意味します
      試す {
          //ロックを解除する
          内部的にロックを解除します(lockData.lockPath);
      ついに
          //マップから削除
          スレッドデータを削除します。(現在のスレッド)
      }
  } 
 
  void releaseLock(String lockPath) 例外をスローします {
      取り消し可能に設定します( null ) ;
      //内部的に保証を使用して、バックグラウンドでノードの削除を試行し続けます
      パスを削除します(パスをロックします);
  }

再入回数が 0 より大きい場合は、再入回数を減らします。 0 に減少すると、zk が呼び出されてノードが削除されます。これは、Redisson 再入可能ロックの解放と一致します。

4. 群集効果

ここでは、分散ロックシナリオで群れ効果を実装するためにZookeeperを使用する方法について説明します。

群集効果とは何か

まず第一に、羊の群れは非常に散在した組織であり、目的もなく、管理も不十分であるため、一般的には飼い主が群れを管理するのを助ける牧羊犬が必要になります。

ある時点で、羊の一匹が前方にもっとおいしい草があることに気づいて動き始めると、周囲の状況に関係なく、残りの羊も突進するようになります。

したがって、群集効果とは、一人の人の合理的な行動が他の人を盲目的に追随させ、非合理的な群集行動を引き起こす状況を指します。

Zookeeper の herd 効果とは、znode が変更された後、回避できたはずの大量の監視通知がトリガーされ、クラスター リソースが無駄になるという事実を指します。

ロックを取得できない場合は進化を待つ

しばらく寝る

スレッドがロックの取得に失敗した場合、しばらくスリープしてから再度ロックの取得を試行することができます。

しかし、この方法は非常に非効率的です。

スリープ時間が短いと、ポーリングが頻繁に実行され、リソースが浪費されます。

スリープ時間が長いと、ロックが解除されたのに取得できないという困った状況になります。

したがって、ここでの最適化のポイントは、アクティブ ポーリングを非同期通知にどのように変更するかということです。

ロックされたノードを監視する

すべてのクライアントがロックを取得したい場合、同じ名前のノードのみを作成します。

znode が存在する場合、これらのクライアントはそこにリスナーを設定します。 znode が削除されると、ロックを待機しているすべてのクライアントに通知され、これらのクライアントは再度ロックを取得しようとします。

ここでは非同期通知のために監視メカニズムが使用されていますが、クライアントの数が特に多い場合はパフォーマンスが低下します。

znode が削除されると、その時点で多数のクライアントに通知を送信する必要があります。この期間中、zk に送信された他の通常のリクエストは遅延またはブロックされる可能性があります。

これにより、単一のポイントの変更 (znode の削除) が包括的な影響 (多数のクライアントへの通知) を引き起こす、集団効果が発生します。

したがって、ここでの最適化のポイントは、znode の監視の数をいかに減らすかであり、最良のケースは 1 つだけです。

前の順序のノードを見る

最初に basePath が指定されている場合、ロックを取得しようとするクライアントは、このパスの下に一時的な順序付きノードを直接作成できます。

作成されたノードが最小ノードである場合、ロックが取得されたことを意味します。最小のノードでない場合は、前のノードに対してのみリスナーを設定し、前のノードの削除動作のみをリッスンします。

この方法では、前のノードが削除されると、すべてのクライアントではなく、次のノードによって表されるクライアントにのみ通知が送信されるため、集団効果が回避されます。

群集効果を回避しながら、現在のロックは公平なロックになります。つまり、スレッドが過度に不足するのを避けるために、ロックは適用された順序で取得されます。

5. 追記

この記事では、ソースコードの観点から Curator を使用して分散ロックを取得するプロセスについて説明し、次にロックを待つ進化プロセスの観点から、分散ロック シナリオで群集効果を回避するための Zookeeper のソリューションを分析します。

これは Zookeeper シリーズの 2 番目の記事です。監視原理と ZAB プロトコルの分析に関する記事も今後公開される予定です。