ZK (ZooKeeper) 分散ロック実装

[[430832]]

準備する

この記事では、事前にセットアップできる 3 つの独立したサーバーを使用します。

構築方法が分からない場合は、以前のZooKeeperクラスタ構築を読んでください：ZooKeeperクラスタの展開について

基本的な ZooKeeper コマンドについては、こちらの記事をご覧ください: ZooKeeper 入門

序文

通常、単一のサービスに対してロックを使用する場合、Java に付属するいくつかのロックを使用して、リソースへの順次アクセスを実装できます。しかし、ビジネスの発展に伴い、基本的にどの企業も複数のサービスを持つようになりました。単純なロックまたは同期では、単一の JVM スレッドの問題しか解決できません。そうすると、単一のサービスに対する Java ロックでは、ビジネスのニーズを満たすことができなくなります。複数のサービスがサービス間で共有リソースにアクセスする問題を解決するために、分散ロックが存在します。分散ロックが生成される理由はクラスタリングです。

文章

分散ロックを実装する方法は何ですか?

分散ロックは主に、ZooKeeper、Reids、Mysql の 3 つの方法で実装されます。

今日は主に、ZooKeeper を使用して分散ロックを実装する方法について説明します。 ZooKeeper のアプリケーションシナリオには、主に次の側面が含まれます。

サービス登録とサブスクリプション（共有ノード）
分散通知（ZNode の監視）
サービスコマンド (ZNode 機能)
データの購読と公開 (ウォッチャー)
分散ロック（一時ノード）

ZooKeeper は、主に強力なデータ一貫性を保証するため、分散ロックを実装します。ロックサービスは次の 2 つのカテゴリに分けられます。

独占的に

現在のロックを取得しようとするすべてのクライアントのうち、現在のロックのキーを正常に取得できるのは 1 つのクライアントだけです。通常、ZooKeeper 上のノード (ZNode) をロックとみなし、一時的なノードを作成することで実装します。複数のクライアントがロックを作成しようとした場合、ロックを正常に作成したクライアントだけがロックを所有できます。

タイミングを制御する

ロックを取得しようとするすべてのクライアントは順番に実行されますが、シーケンス番号 (zxid) が付きます。たとえば、/testLock というノードがあり、すべての一時ノードはこのノードの下に作成されます。 ZK の親ノード (/testLock) はシーケンス番号を維持します。これは ZK のプロパティです。これにより、子ノードが順番に作成され、各クライアントのグローバルシーケンスが形成されます。

ZKロック機構

ZooKeeper 分散ロックを実装する前に、ZooKeeper 分散ロックメカニズムの実装プロセスと原則を理解する必要があります。そうでなければ、面接に行ったときにどうやって面接官と話せばいいのでしょうか?

一時シーケンスノード

ZooKeeper の一時的なシーケンシャルノードに基づいて、ZooKeeper は分散ロックの実装に適しています。

シーケンス番号ジェネレーター: ZooKeeper の各ノードには、シーケンスジェネレーターが組み込まれています。各ノードの下に一時ノードが作成され、新しい子ノードの後にシーケンス番号が追加されます。この生成された数値は、前の数値に 1 加算されます。
順序付けられた増分: ZooKeeper ノードは順序付けられた増分であり、ロックの公平性を確保できます。永続的な親ノードの下に対応する一時シーケンスノードを作成するだけで済みます。各スレッドがロックを占有しようとする前に、watch を呼び出して、現在のシーケンス番号が現在の親ノード内で最小であるかどうかを判断します。そうであれば、ロックを取得します。
Znode 監視: 各スレッドは、ロックをプリエンプトする前に、現在のスレッドに属する ZNode ノードを作成します。ロックを解除すると、作成された ZNode は削除されます。作成したシーケンス番号が最小でない場合は、監視通知、つまり前の ZNode のステータス通知を待機します。現在の ZNode が削除されると、コールバックメカニズムがトリガーされ、次の ZNode にロックを取得して作業を開始できることが通知されます。
一時ノードの自動削除: ZooKeeper にはもう一つの利点があります。クライアントが切断されると、作成した一時ノードは自動的に削除されます。したがって、分散ロックを使用する場合、通常はネットワーク異常やその他の理由によるデッドロックを回避するために一時的なノードを作成します。
群集効果: ZooKeeper ノードの順次アクセス可能性と、前方のノードを後方で監視する方法により、群集効果を効果的に回避できます。群集効果とは何か: ノードがハングアップすると、すべてのノードが監視して応答する必要があり、サーバーに大きな負担がかかります。一時的に連続するノードがある場合、ノードがハングアップすると、その後ろのノードのみが応答します。

次の図を見てみましょう。

上図では、ZooKeeper にロックノード testLock があります。このロックは ZooKeeper のノードです。 2 つのクライアントがこのロックを取得する場合、ZooKeeper にロックを要求します。これを一時的なシーケンシャルノードと呼びます。

/testLock ディレクトリに一時的なシーケンシャルノードを作成すると、ZK はこの一時的なノードのノードシリアル番号を自動的に維持し、このノードが増分されます。たとえば、clientA が一時的なシーケンシャルノードを作成すると、ZK はシリアル番号 /lock0000000001 を生成します。次に、clientB も一時的なシーケンシャルノードを生成し、ZK はシリアル番号 /lock0000000002 を生成します。ここで、数字は 1 から順番に増加し、ZK は内部的にこの順序を維持します。

下の図に示すように:

このとき、ClientA は、親ノードの下で自分が最小であるかどうかを監視して判断します。もしそうなら、私は一番小さくて他の人は私より大きいので、ロックすることができます。自分でロックできます。あなたはすでに成熟した一時ノードであり、それを自分でロックする方法を学ぶ必要があります。さて、ZKはどのように判断するのでしょうか?ハニー、続きを読んでください：

これはcleintAがロックを完了したということです。この時点で、clientB もロックしたいので、/testLock に独自の一時ノードを作成する必要があります。その場合、次の図に示すように、clientB のシリアル番号は /lock0000000002 になります。

このとき、先ほど言ったことが出てきます。 clientB がロックすると、それが最小かどうかを判断します。現在の親ノードの下では最小ではないことがわかります。あ～私って結構大きいし、私より小さいのもいるんですよ！！！

ロックに失敗しました。このとき、クライアントBはクライアントAを覗き見します。だんだんと雰囲気が曖昧になってきます。いいえ、作業が完了したかどうかを確認するために、前のノード (clientA) を監視するためです。完了すると、クライアントBはロック作業を実行できます。ベイビー、下の写真を見てください。

クライアント A が正常にロックされると、クライアント A は独自の業務を処理します。クライアント A は作業を終了すると、「完了しました。次に進みます。」と言います。それで、クライアントAはどうやって仕事を終えたのでしょうか?どれくらい時間がかかりましたか？いいえ、具体的なプロセスは何ですか?小農、君は間違っている。あなたは何について話しているのですか！！！とても恥ずかしい

上記で、clientB がロックに失敗すると、リスナーが前のノード (clientA) に追加されると述べていませんでしたか。 clientA が削除されると、誰かがロックを解除したことを意味し、この時点で clientB にロックを再度取得するように通知されます。

このとき、clientB がロックを再取得すると、現在の親ノードの下で最小のロックであることがわかり、clientB はロックを開始して作業を開始し、一連の操作を実行します。クライアント B が終了すると、ロックが解除され、「次」と表示されます。

次の図に示すように:

もちろん、clientA、clientB 以外にも、C\D\E なども存在します。これらの文字は奇妙で馴染みのあるように見えますが、原理は同じです。最小のノードがロック解除されます。そうでない場合は、前のノードが解放されるかどうかを監視します。解除された場合は再度ロックを試みてください。前のセクションのノードが解放されると、そのノードは最小になり、キューの先頭に配置されます。これは、銀行で番号を取得する操作に少し似ています。

コードの実装

ZooKeeper を使用して一時シーケンスノードを作成し、分散ロックを実装します。一般的なプロセスは、最初に永続的な親ノードを作成し、現在のノードの下に一時的なシーケンスノードを作成し、最小のシーケンス番号を見つけて分散ロックを取得し、プログラム業務が完了したらロックを解除し、次のノードに操作を通知することです。 watch を使用してノードの変更を監視し、次に次に小さいシーケンスノードに対して順番に操作を行います。

まず、永続的な親ノードを作成する必要があります。私の場合は /mxn です。

ウォッチコールバック

org.apache.zookeeper.* をインポートします。
 org.apache.zookeeper.data.Stat をインポートします。 
 
 java.util.Collections をインポートします。
 java.util.List をインポートします。
 java.util.concurrent.CountDownLatch をインポートします。 
 
 
 /**
 * @プログラム: mxnzookeeper
 * @ClassName ウォッチコールバック
 * @説明：
 * @author: WeChat検索: 穆小農
 * @作成: 2021-10-23 10:48
 * @バージョン 1.0
 **/
パブリッククラス WatchCallBack は Watcher、AsyncCallback.StringCallback、AsyncCallback.Children2Callback、AsyncCallback.StatCallback を実装します { 
 
    ZooKeeper zk ;
    文字列スレッド名;
    カウントダウンラッチcc = 新しいカウントダウンラッチ(1);
    文字列パス名; 
 
パブリック文字列 getPathName() {
パス名を返します。
    } 
 
パブリックvoid setPathName(文字列 pathName) {
        this.pathName = パス名;
    } 
 
パブリック文字列 getThreadName() {
スレッド名を返します。
    } 
 
パブリックvoid setThreadName(String threadName) {
        this.threadName = スレッド名;
    } 
 
パブリックZooKeeper getZk() {
 zkを返します。
    } 
 
パブリックvoid setZk(ZooKeeper zk) {
        this.zk = zk;
    } 
 
    /** @Author 牧小农
     * @Description //TODO メソッドをロックしてみる
     * @日付16:14 2021/10/24
     * @パラメータ
     * @戻る  
     **/
パブリックボイドtryLock(){
        試す { 
 
            システム。 out .println(threadName + " 作成を開始します..." );
            // 連続した一時ノードを作成する
            そうです。作成( "/lock"、threadName.getBytes()、ZooDefs.Ids.OPEN_ACL_UNSAFE、CreateMode.EPHEMERAL_SEQUENTIAL、this、"abc " ) ;
            //現在のノードをブロックし、前のノードがロックを解除するかどうかを監視する
            cc.await();
        } キャッチ (InterruptedException e) {
            e.printStackTrace();
        }
    } 
 
    /** @Author 牧小农
     * @Description //TODO ロック解除方法
     * @日付16:14 2021/10/24
     * @パラメータ
     * @戻る  
     **/
パブリックボイドロック解除(){
        試す {
            //ロックを解除し、一時ノードを削除します
            zk.delete (パス名、-1);
            //作業終了
            システム。 out .println(threadName + " 作業が完了しました...." );
        } キャッチ (InterruptedException e) {
            e.printStackTrace();
        } キャッチ (KeeperException e) {
            e.printStackTrace();
        }
    } 
 
 
    @オーバーライド
パブリックvoid プロセス(WatchedEvent イベント) { 
 
        //最初のノードがロックを解除すると、2番目のノードはコールバックを受け取ります
        //前のノードが解放されたことを伝え、ロックの取得を開始できます
        スイッチ（イベント.getType()）{
ケースなし:
                壊す;
 NodeCreatedの場合:
                壊す;
 NodeDeletedの場合:
                //現在のノードがロックを再取得します
                zk.getChildren( "/" 、 false 、 this 、 "sdf" );
                壊す;
 NodeDataChangedの場合:
                壊す;
 NodeChildrenChangedの場合:
                壊す;
        } 
 
    } 
 
    @オーバーライド
パブリックvoid processResult( int rc, 文字列パス, オブジェクト ctx, 文字列名) {
        if(名前!= null ){
            システム。 out .println(threadName + "スレッドがノードを作成しました: " + name );
            パス名 =名前;
            //前のノードを監視する
            zk.getChildren( "/" 、 false 、 this 、 "sdf" );
        } 
 
    } 
 
    //getChildrenコールバック
    @オーバーライド
パブリックvoid processResult( int rc、String path、Object ctx、List<String> children、Stat stat) { 
 
        // ノードは番号に従って昇順に並べられます
        コレクション.sort(子供);
        //ノードをインターセプトします。たとえば、/lock0000000022 はインターセプト後に lock0000000022 になります。
 int i = children.indexOf(pathName.部分文字列(1)); 
 
 
        //最初のものですか、つまり一番小さいものですか
        もし(i == 0){
            // は最初の
            システム。 out .println(threadName + " 今、私は末っ子です...." );
            試す {
                zk.setData( "/" 、threadName.getBytes()、-1);
                cc.countDown(); 
 
            } キャッチ (KeeperException e) {
                e.printStackTrace();
            } キャッチ (InterruptedException e) {
                e.printStackTrace();
            }
        }それ以外{
            //最初のものではない
            //前のノードを監視して、作業が完了してロックが解放されたかどうかを確認します
            zk.exists( "/" +children.get(i-1),this,this, "sdf" );
        } 
 
    } 
 
    @オーバーライド
パブリックvoid processResult( int rc, String path, Object ctx, Stat stat) {
        //障害が存在するかどうかを判断する
    }
 }

テストロック

com.mxn.zookeeper.config.ZKUtils をインポートします。
 org.apache.zookeeper.ZooKeeper をインポートします。
 org.junit.Afterをインポートします。
 org.junit.Before をインポートします。
 org.junit.Test をインポートします。 
 
 
 /**
 * @プログラム: mxnzookeeper
 * @ClassName テストロック
 * @説明：
 * @author: WeChat検索: 穆小農
 * @作成: 2021-10-23 10:45
 * @バージョン 1.0
 **/
パブリッククラスTestLock{ 
 
 
    ZooKeeper zk ; 
 
    @前に
パブリックボイドconn（）{
        ZKUtils の getZK() メソッドは、次のコードで使用できます。
    } 
 
    @後 
パブリックボイドクローズ（）{
        試す {
            zk.close ();
        } キャッチ (InterruptedException e) {
            e.printStackTrace();
        }
    } 
 
    @テスト
パブリックボイドロック（）{ 
 
        // 10個のスレッドを作成する
( int i = 0; i < 10; i++) {
            新しいスレッド(){
                @オーバーライド
パブリックボイド実行（）{
                    WatchCallBack watchCallBack = 新しい WatchCallBack();
                    watchCallBack.setZk(zk);
                    文字列 threadName = Thread.currentThread().getName();
                    watchCallBack.setThreadName(スレッド名);
                    //スレッドはロック取得操作を実行します
                    watchCallBack.tryLock();
                    試す {
                        //ビジネスロジック処理を実行する
                        システム。 out .println(threadName+ "ビジネス ロジックの処理を開始します..." );
                        スレッド.sleep(200);
                    }(例外 e) をキャッチ{
                        e.printStackTrace();
                    }
                    //ロックを解除する
                    watchCallBack.unLock(); 
 
 
                }
            }。始める（）;
        } 
 
 
        while( true ){ 
 
        } 
 
    } 
 
 }

実行結果:

スレッド 1 スレッドがノードを作成しました: /lock0000000112
スレッド 5 スレッドがノードを作成しました: /lock0000000113
スレッド 2 スレッドがノードを作成しました: /lock0000000114
スレッド6はノードを作成します: /lock0000000115
スレッド 9 スレッドがノードを作成しました: /lock0000000116
スレッド 4 スレッドがノードを作成しました: /lock0000000117
スレッド 7 スレッドがノードを作成しました: /lock0000000118
スレッド3はノードを作成します: /lock0000000119
スレッド 8 スレッドがノードを作成しました: /lock0000000120
スレッド 0 スレッドがノードを作成しました: /lock0000000121
スレッド-1 今、私は末っ子です…。
スレッド 1 はビジネス ロジックの処理を開始します...
スレッド 1 は作業を終了しました。
スレッド-5 今、私は末っ子です…。
スレッド 5 はビジネス ロジックの処理を開始します...
スレッド5は作業を終了しました...
スレッド-2 今私は末っ子です…。
スレッド 2 はビジネス ロジックの処理を開始します...
スレッド2 作業完了しました。
スレッド-6 今、私は末っ子です....
スレッド 6 はビジネス ロジックの処理を開始します...
スレッド6 作業完了しました。
スレッド-9 今、私は末っ子です....
スレッド 9 はビジネス ロジックの処理を開始します...
スレッド9は作業を終了しました...
スレッド-4 今、私は末っ子です…。
スレッド 4 はビジネス ロジックの処理を開始します...
スレッド4 作業完了しました...
スレッド-7 今、私は末っ子です....
スレッド 7 はビジネス ロジックの処理を開始します...
スレッド7は作業を終了しました。
スレッド-3 今、私は末っ子です....
スレッド 3 はビジネス ロジックの処理を開始します...
スレッド3 作業完了しました。
スレッド-8 今、私は末っ子です....
スレッド 8 はビジネス ロジックの処理を開始します...
スレッド8は作業を終了しました。
スレッド-0 今私は末っ子です…。
スレッド 0 はビジネス ロジックの処理を開始します...
スレッド 0 は作業を終了しました。