Zookeeper に基づく分散ロック

この記事を読むのに必要な時間はたった 10 分です。

現在、分散ロックを実装するための一般的なソリューションは、データベース、Redis、Zookeeper に基づくソリューションの 3 つです。最初の 2 つの解決策については、インターネット上に参考になる情報が多数ありますが、この記事では詳しく説明しません。 Zookeeper を使用して分散ロックを実装する方法を見てみましょう。

Zookeeper とは何ですか?

Zookeeper (業界では zk と略される) は、構成管理、分散コラボレーション、および命名機能を提供する集中型サービスです。これらの機能は、分散システムにおいて非常に低レベルかつ不可欠な基本機能です。しかし、これらの機能を自社で実装し、高スループット、低レイテンシ、一貫性、可用性を実現するのは実際には非常に困難です。したがって、Zookeeper はこれらの機能を提供しており、開発者は Zookeeper 上に独自の分散システムを構築します。

ZooKeeper の実装は比較的複雑ですが、ZooKeeper が提供するモデルの抽象化は非常にシンプルです。 Zookeeper は、複数レベルのノード名前空間 (ノードは znode と呼ばれます) を提供します。各ノードはスラッシュ (/) で区切られたパスで表され、各ノードには親ノード (ルート ノードを除く) があり、これはファイル システムに非常に似ています。たとえば、/foo/doo は、親ノードが /foo で、親ノードが / である znode を表します。また、/ はルート ノードであり、親ノードはありません。ファイル システムとは異なり、これらのノードはすべて関連データで設定できますが、ファイル システムではファイル ノードのみがデータを保存でき、ディレクトリ ノードはデータを保存できません。高いスループットと低いレイテンシを確保するために、Zookeeper はこのツリー状のディレクトリ構造をメモリ内に維持します。この機能により、Zookeeper は大量のデータの保存に使用できなくなります。各ノードのデータ保存上限は1Mです。

高可用性を確保するには、Zookeeper をクラスター形式で展開する必要があります。これにより、クラスター内のほとんどのマシンが利用可能である限り (特定のマシン障害を許容できる)、Zookeeper 自体も引き続き利用可能になります。 Zookeeper を使用する場合、クライアントはクラスター マシンのリストを認識し、クラスター内のマシンとの TCP 接続を確立してサービスを使用する必要があります。クライアントはこの TCP 接続を使用して、要求の送信、結果の取得、監視イベントの取得、ハートビート パケットの送信を行います。この接続が異常切断された場合、クライアントは別のマシンに接続できます。

簡略化されたアーキテクチャ図は次のとおりです。

クライアントの読み取り要求は、クラスター内の任意のマシンで処理できます。読み取り要求によってノードにリスナーが登録されると、そのリスナーも接続された Zookeeper マシンによって処理されます。書き込み要求の場合、これらの要求は同時に他の Zookeeper マシンに送信され、合意に達した後にのみ要求が成功として返されます。したがって、ZooKeeper クラスター マシンの数が増えると、読み取り要求のスループットは増加しますが、書き込み要求のスループットは減少します。

順序は Zookeeper にとって非常に重要な機能です。すべての更新はグローバルに順序付けられ、各更新には zxid (Zookeeper Transaction Id) と呼ばれる一意のタイムスタンプが付きます。読み取り要求は更新に対してのみ順序付けられます。つまり、読み取り要求の返される結果には、この Zookeeper*** の zxid が含まれます。

Zookeeper を使用して分散ロックを実装するにはどうすればよいですか?

アルゴリズムのフローを説明する前に、ZooKeeper のノードの興味深いプロパティをいくつか見てみましょう。

• 順序付けられたノード:現在親ノード /lock が存在する場合、この親ノードの下に子ノードを作成できます。 zookeeper はオプションの順序付き機能を提供します。たとえば、子ノード「/lock/node-」を作成し、ordered を指定できます。その後、Zookeeper は子ノードを生成するときに、現在の子ノードの数に応じて整数のシリアル番号を自動的に追加します。つまり、最初に作成された子ノードの場合、生成される子ノードは /lock/node-0000000000 となり、次のノードは /lock/node-00000000001 となり、以下同様に続きます。
• 一時ノード:クライアントは一時ノードを作成できます。セッションが終了するかタイムアウトすると、Zookeeper は自動的にノードを削除します。
• イベント監視:データを読み取るときに、ノードのイベント監視を設定することもできます。ノードのデータまたは構造が変更されると、Zookeeper はクライアントに通知します。現在、Zookeeper には次の 4 つのイベントがあります: 1) ノードの作成。 2) ノードの削除3) ノードデータの変更4) 子ノードの変更。

以下は、ロック スペースのルート ノードが /lock であると仮定して、Zookeeper を使用して分散ロックを実装するアルゴリズム フローを説明します。

• クライアントは Zookeeper に接続し、/lock の下に一時的かつ順序付けられた子ノードを作成します。最初のクライアントに対応する子ノードは /lock/lock-0000000000、2 番目のクライアントに対応する子ノードは /lock/lock-0000000001 などとなります。
• クライアントは /lock の下の子ノード リストを取得し、作成した子ノードが現在の子ノード リスト内で最小のシーケンス番号を持つ子ノードであるかどうかを判断します。そうであれば、ロックを取得したとみなされます。それ以外の場合は、/lock の子ノード変更メッセージを監視し、子ノード変更通知を取得してからロックが取得されるまでこの手順を繰り返します。
• ビジネス コードを実行します。
• 業務プロセスが完了したら、対応する子ノードを削除してロックを解除します。

ステップ 1 で作成された一時ノードにより、障害が発生した場合でもロックを解除できるようになります。次のシナリオを考えてみましょう。クライアント a によって現在作成されている子ノードがシリアル番号が最も小さいノードである場合、クライアントが配置されているマシンはロックを取得した後にクラッシュし、クライアントは子ノードを積極的に削除しません。最大のシリアル番号を持つノードが作成されると、ロックは解放されず、デッドロックが発生します。一時ノードが作成されるため、クライアントがクラッシュした後、一定時間が経過すると、Zookeeper はクライアントのハートビート パケットを受信できなくなり、セッションが無効であると判断し、一時ノードを削除してロックを解除します。

さらに、注意深い友人は、ステップ 2 で子ノード リストを取得し、リスナーを設定する際の原子性の問題を考えるかもしれません。次のようなシナリオを考えてみましょう。クライアント a の対応する子ノードは /lock/lock-0000000000 であり、クライアント b の対応する子ノードは /lock/lock-00000000001 です。クライアント 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、2 番目のクライアントに対応する子ノードは /lock/lock-0000000001 などとなります。
• クライアントは /lock の下の子ノードのリストを取得し、作成した子ノードが現在の子ノード リスト内で最小のシーケンス番号を持つ子ノードであるかどうかを判断します。そうであれば、ロックが取得されたとみなします。それ以外の場合は、直前の子ノードの削除メッセージをリッスンします。子ノードの変更通知を受信した後、ロックが取得されるまでこの手順を繰り返します。
• ビジネス コードを実行します。
• 業務プロセスが完了したら、対応する子ノードを削除してロックを解除します。

Curatorのソースコード解析

Zookeeper ネイティブ クライアントによって公開される API は非常に簡潔ですが、分散ロックを実装するのは依然として面倒です... Curator オープン ソース プロジェクトによって提供される Zookeeper 分散ロック実装を直接使用できます。

次のパッケージ (Maven ベース) を導入するだけです。

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

そうすれば使えますよ！コードは次のとおりです。

公共 静的void main(String[] args)は例外をスローします{
    // Zookeeperクライアントを作成する
    再試行ポリシー retryPolicy = 新しい ExponentialBackoffRetry(1000, 3);
    CuratorFramework クライアント = CuratorFrameworkFactory.newClient( "10.21.41.181:2181,10.21.42.47:2181,10.21.49.252:2181" 、 retryPolicy);
    クライアントを起動します。 
 
    //分散ロックを作成します。ロックスペースのルートノードパスは/curator/lockです。
    InterProcessMutex ミューテックス = 新しい InterProcessMutex(client, "/curator/lock" );
    ミューテックスを取得します。
    //ロックを取得してビジネスプロセスを続行します
    システム。 out .println( "ミューテックスに入る" );
    //ビジネスプロセスを完了し、ロックを解除する
    ミューテックスを解放します。 
     
    //クライアントを閉じる
    クライアントを閉じます() ;
 }

主要なコア操作は mutex.acquire() と mutex.release() だけであることがわかります。これは非常に便利です。

ロックを取得するソースコードの実装を分析してみましょう。取得方法は以下の通りです。

 /*
 * ロックを取得します。ロックが占有されている場合は、ブロックされて待機します。この操作は、同じスレッドの再入可能性 (つまり、ロックの繰り返し取得) をサポートします。取得数はリリース数と同じである必要があります。
 * @throws Exception ZKエラー、接続中断
 */
 @オーバーライド
public void acquire() は例外をスローします
{
    if ( !internalLock(-1, null ) )
    {
        throw new IOException( "ロックを取得中に接続が失われました: " + basePath);
    }
 }

ここで注意すべき点が1つあります。 Zookeeper との通信で例外が発生した場合、 acquire は直接例外をスローするため、ユーザーは再試行戦略を自分で作成する必要があります。このコードは internalLock(-1, null) を呼び出し、パラメータはロックがブロックされ、占有されているときに待機することを示します。 internalLock のコードは次のとおりです。

 private boolean internalLock(long time , TimeUnit unit) は例外をスローします
{ 
 
    //ここでは同じスレッドの再入可能性を処理します。ロックが取得された場合は、対応するデータ構造内の取得カウントを増やし、直接成功を返します。
    スレッド currentThread = Thread.currentThread();
    ロックデータ lockData = threadData.get(currentThread);
    ロックデータ != nullの場合
    {
        // 再入力
        ロックデータ.ロックカウント.増加および取得();
戻る 真実;
    } 
 
    //ここで実際にZookeeperからロックを取得します
    文字列 lockPath = internals.attemptLock( time , unit, getLockNodeBytes());
    lockPath がnullの場合
    {
        //ロックを取得したら、ロックを取得している現在のスレッドの情報を記録します。再入力する場合は、LockDataの回数を増やすだけです
        LockData newLockData = 新しい LockData(currentThread、lockPath);
        threadData.put(現在のスレッド、新しいロックデータ);
戻る 真実;
    } 
 
    //ブロッキングが返されるときにロックを取得できません。ここでのコンテキスト処理は、Zookeeper 通信が異常であることを意味します。
戻る 間違い;
 }

特定のコメントがコードに追加され、展開されません。 Zookeeper がロックを取得する具体的な実装を見てみましょう。

文字列 attemptLock(long time , TimeUnit unit, byte[] lockNodeBytes) は例外をスローします
{
    //パラメータの初期化、ここでは省略
    //... 
    
    // スピンしてロックを取得する
    while ( !isDone )
    {
        完了 = true ; 
 
        試す
        {
            //ロック空間の下に一時的かつ順序付けられた子ノードを作成する
            ourPath = driver.createsTheLock(クライアント、パス、localLockNodeBytes);
            //ロックが取得されたかどうかを判断します（子ノードのシーケンス番号が最も小さい）。ロックが取得された場合は直接戻り、そうでない場合はブロックして前の子ノードの削除通知を待ちます。
            ロックがある = internalLockLoop(startMillis、millisToWait、ourPath);
        }
        キャッチ (KeeperException.NoNodeException e)
        {
            //NoNodeException の場合、コードはセッションの有効期限が切れたときにのみ NoNodeException がスローされることを保証するため、再試行戦略に従って再試行がここで実行されます。
            if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
            {
                isDone = false ;
            }
それ以外 
            {
                eを投げる;
            }
        }
    } 
 
    //ロックが取得された場合、子ノードのパスが返されます
    if ( ロックがある )
    {
 ourPathを返します。
    } 
 
戻る ヌル;
 }

上記のコードには主に 2 つのステップがあります。

• driver.createsTheLock: 一時的かつ順序付けられた子ノードを作成します。実装は比較的単純であり、拡張されることはありません。いくつかのノード モードに焦点を当てます: 1) PERSISTENT (***); 2) PERSISTENT_SEQUENTIAL (*** かつ順序付き) 3) EPHEMERAL（一時的） 4) EPHEMERAL_SEQUENTIAL (一時的かつ順序付き)。
• internalLockLoop: ロックが取得されるまでブロックして待機します。

internalLockLoop がロックとブロッキング待機をどのように決定するかを見てみましょう。ここでは無関係なコードが削除され、メイン プロセスのみが残ります。

 //ロックが取得されるまでスピンする
( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) の間
{
    //すべての子ノードのリストを取得し、昇順で並べ替えます
    リスト<String> children = getSortedChildren(); 
     
    //現在の子ノードがシーケンス番号に従って最小の子ノードであるかどうかを判断します
    文字列sequenceNodeName = ourPath。部分文字列(basePath.length() + 1); //スラッシュを含めるには+1
    述語結果 predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
    if ( predicateResults.getsTheLock() )
    {
        //最小の子ノードであれば、ロックを取得したとみなされます
        ロックを有効にする = true ;
    }
それ以外 
    {
        //それ以外の場合は前の子ノードを取得します
        文字列 previousSequencePath = basePath + "/" + predicateResults.getPathToWatch(); 
 
        //ここでは、スレッドの同期にオブジェクト モニターを使用します。ロックを取得できない場合は、前の子ノードの削除メッセージをリッスンし、wait() を実行します。前の子ノードが削除されると (つまり、ロックが解放されると)、コールバックはnotifyAllを介してこのスレッドを起動し、このスレッドはロックが取得されたかどうかを判断するためにスピンし続けます。
        同期(これ)
        {
            試す
            {
                // 前の子ノードが削除されている場合は例外がスローされ、イベント リスナーが設定されないため、ここでは checkExists() ではなく getData() インターフェイスが使用されます。 checkExists はノードが存在するかどうかの情報も取得できますが、トリガーされることのないリスナーも設定するため、Zookeeper のリソース リークが発生します。
                client.getData().usingWatcher(watcher).forPath(前のシーケンスパス); 
 
                //ブロッキング待機時間を設定する場合
                if ( millisToWait != null )
                {
                    millisToWait -= (System.currentTimeMillis() - startMillis);
                    startMillis = System.currentTimeMillis();
                    if ( ミリ秒待機時間 <= 0 )
                    {
                        削除する = true ; // 待機時間が来たら、対応する子ノードを削除します
                        壊す;
                    } 
                     
                    //適切な時間まで待つ
                    待機(ミリ秒単位の待機)。
                }
それ以外 
                {
                   //永遠に待つ
                    待って（）;
                }
            }
            キャッチ (KeeperException.NoNodeException e)
            {
                // getData を使用して上記のリスナーを設定する場合、前の子ノードが削除されていると、NoNodeException がスローされます。一度スピンするだけで、追加の処理は必要ありません。
            }
        }
    }
 }

具体的なロジックはコメントに示されており、ここでは繰り返さないことにします。コードに設定されたイベント リスナーは、イベント コールバックが発生したときに現在のスレッドを起動して再スピンするように notifiesAll するだけです。比較的単純なので拡張されません。

その上。