私は分散ロックについて何度も読み返してきました。最善の解決策をお教えします

[[392389]]

分散ロックのシナリオ

フラッシュセールのシナリオケース

フラッシュセールでは、在庫の過剰販売や値引きの繰り返しなどの同時発生の問題を防ぐ必要があります。通常、共有リソースの競合によって生じるデータの不整合の問題を解決するには、分散ロックを使用する必要があります。

携帯電話のフラッシュセールのシナリオを例にとると、ラッシュ購入のプロセスには通常 3 つのステップがあります。

1. 該当する製品の在庫を減算します。 2. 製品の注文を作成します。 3. ユーザーが支払います。

このようなシナリオでは、分散ロックを使用して問題を解決できます。たとえば、ユーザーがフラッシュセールの「注文」リンクを入力すると、商品の在庫をロックし、操作が完了した後に在庫の減額などの操作を完了することができます。ロックを解除すると、次のユーザーが引き続き入ることができ、在庫のセキュリティが確保されます。また、フラッシュセールの失敗による DB ロールバックの数も削減できます。全体のプロセスを下の図に示します。

注: ロックの粒度は、特定のシナリオと要件に応じて検討する必要があります。

3種類の分散ロック

Zookeeper の分散ロック実装では、主に Zookeeper の 2 つの機能が使用されます。

Zookeeperノードは繰り返し作成できません
Zookeeper の Watcher 監視メカニズム

不公平なロック

不公平なロックの場合、ロックのプロセスを次の図に示します。

長所と短所

実際、上記の実装には利点と欠点があります。

アドバンテージ：

実装は比較的単純で、通知メカニズムを備えており、より高速な応答を提供できます。これはReentrantLockの考え方に少し似ています。ノードの削除が失敗するシナリオでは、セッションタイムアウトによってノードを削除できることが保証されます。

欠点:

重量が重いため、ロックの数が多いと「群れを混乱させる」という問題が発生する可能性があります。

「群れを怖がらせる」とは、ノードが削除されると、このノードの削除アクションのために Watcher にサブスクライブしている多数のスレッドがコールバックすることを意味し、これは Zk クラスターにとって非常に有害です。したがって、この現象を回避する必要があります。

「ショック集団」の解決:

「雷鳴の群れ」問題を解決するには、ノードをサブスクライブする戦略を放棄する必要がありますが、どうすればよいでしょうか?

ロックをディレクトリに抽象化し、複数のスレッドがこのディレクトリの下に瞬時に連続ノードを作成します。 Zookeeper がノードの順序を保証するため、ノードの順序を使用してロックを判断できます。
まず、連続ノードを作成し、次に現在のディレクトリ内の最小のノードを取得し、その最小のノードが現在のノードであるかどうかを判断します。そうであれば、ロックは正常に取得されます。そうでない場合、ロックの取得は失敗します。
ロックの取得に失敗したノードは、現在のノードの前の連続ノードを取得し、このノードのリスナーを登録し、ノードが削除されたときに現在のノードに通知します。
ロック解除時はノードを削除した後に次のノードに通知されます。

フェアロック

不公平なロックの欠点に基づいて、次の解決策を通じてそれを回避できます。

長所と短所

利点:前述のように、一時的なシーケンシャルノードを使用すると、複数のノードの同時競合ロックを回避でき、サーバーへの負荷を軽減できます。

欠点:読み取り/書き込みシナリオでは、一貫性の問題を解決できません。読み取り中にロックが取得されると、パフォーマンスが低下します。このような問題には、JDK の ReadWriteLock などの読み取り/書き込みロックを使用できます。

読み取り/書き込みロックの実装

読み取り/書き込みロックの特性: 複数のスレッドが読み取りを行っている場合、読み取り/書き込みロックは同時に読み取ることができ、ロックフリーの状態になります。書き込みロックが動作中の場合、読み取りロックは書き込みロックを待機する必要があります。書き込みロックを追加する場合、前の読み取りロックが同時実行されるため、書き込みロックを実行する前に最後の読み取りロックの完了を監視する必要があります。手順は次のとおりです。

読み取り要求。前の要求が読み取りロックの場合は、監視せずに直接読み取ることができます。前に 1 つ以上の書き込みロックがある場合は、最後の書き込みロックのみを監視する必要があります。
書き込み要求の場合、前のノードを監視するだけで済みます。 Watcher メカニズムはミューテックスと同じです。

分散ロックの実践

この記事のソースコードで使用されている環境: JDK 1.8、Zookeeper 3.6.x

キュレーターコンポーネントの実装

POM 依存関係

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

ミューテックスロックアプリケーション

Zookeeper の不公平なロックの「集団ショック」効果により、不公平なロックは実際には Zookeeper における最良の選択ではありません。以下は、Zookeeper 分散ロックを使用するためのスパイクをシミュレートする例です。

パブリッククラスMutexTest {
静的ExecutorService executor = Executors.newFixedThreadPool(8);
静的AtomicIntegerストック = 新しいAtomicInteger(3);
公共 静的void main(String[] args)はInterruptedExceptionをスローします{
        CuratorFrameworkクライアント = getZkClient();
        文字列キー= "/lock/lockId_111/111" ;
        最終的な InterProcessMutex mutex = 新しい InterProcessMutex(client, key );
 ( int i = 0; i < 99; i++) {
            実行者.送信(() -> {
                （stock.get() < 0）の場合{
                    System.err.println( "在庫が不足しています。直接返品してください" );
戻る;
                }
                試す {
                    ブール値 acquire = mutex.acquire(200, TimeUnit.MILLISECONDS);
                    （取得）の場合{
 int s = stock.decrementAndGet();
                        （s < 0）の場合{
                            System.err.println( "フラッシュセール開始、在庫不足" );
                        }それ以外{
                            システム。 out .println( "購入が成功しました。残りの在庫: " + s);
                        }
                    }
                } キャッチ (例外 e) {
                    e.printStackTrace();
                ついに
                    試す {
                        （mutex.isAcquiredInThisProcess()）の場合
                            ミューテックスを解放します。
                    } キャッチ (例外 e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        （真）の間{
            実行者が終了した場合
                実行者.シャットダウン();
                システム。 out .println( "フラッシュセール終了後、残りの在庫は次のとおりです: " + stock.get());
            }
            TimeUnit.MILLISECONDS.sleep(100);
        }
    }
    プライベート静的CuratorFramework getZkClient() {
        文字列 zkServerAddress = "127.0.0.1:2181" ;
        ExponentialBackoffRetry 再試行ポリシー = 新しい ExponentialBackoffRetry(1000, 3, 5000);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerアドレス)
                .セッションタイムアウトMs(5000)
                .接続タイムアウトMs(5000)
                .retryPolicy(再試行ポリシー)
                。建てる（）;
        zkClient を起動します。
 zkClientを返します。
    }
 }

読み取り/書き込みロックアプリケーション

読み取り/書き込みロックは、複数のスレッドによる読み取り時にロックフリーであり、書き込みロックが先行している場合にのみ、書き込みロックが完了するまで待機してからデータにアクセスするため、読み取り/書き込みロックを使用すると、キャッシュの二重書き込みの強力な一貫性を確保できます。

パブリッククラスReadWriteLockTest {
静的ExecutorService executor = Executors.newFixedThreadPool(8);
静的AtomicIntegerストック = 新しいAtomicInteger(3);
静的InterProcessMutex readLock;
静的InterProcessMutex 書き込みロック;
公共 静的void main(String[] args)はInterruptedExceptionをスローします{
        CuratorFrameworkクライアント = getZkClient();
        文字列キー= "/lock/lockId_111/1111" ;
        InterProcessReadWriteLock readWriteLock = 新しい InterProcessReadWriteLock(クライアント、キー);
        readLock は readWriteLock.readLock() です。
        writeLock = readWriteLock.writeLock();
 ( int i = 0; i < 16; i++) {
            実行者.送信(() -> {
                試す {
                    ブール値read = readLock.acquire(2000, TimeUnit.MILLISECONDS);
                    if (読み取り) {
 int num = stock.get();
                        システム。 out .println( "在庫を読み取ります。現在の在庫は: " + num);
                        （数値<0）の場合{
                            System.err.println( "在庫が不足しています。直接返品してください" );
戻る;
                        }
                    }
                } キャッチ (例外 e) {
                    e.printStackTrace();
                }ついに {
                    readLock.isAcquiredInThisProcess() の場合 {
                        試す {
                            読み取りロックを解除します。
                        } キャッチ (例外 e) {
                            e.printStackTrace();
                        }
                    }
                }
                試す {
                    ブール値 acquire = writeLock.acquire(2000, TimeUnit.MILLISECONDS);
                    （取得）の場合{
 int s = stock.get();
                        （s <= 0）の場合{
                            System.err.println( "フラッシュセール開始、在庫不足" );
                        }それ以外{
                            s = stock.decrementAndGet();
                            システム。 out .println( "購入が成功しました。残りの在庫: " + s);
                        }
                    }
                } キャッチ (例外 e) {
                    e.printStackTrace();
                ついに
                    試す {
                        (writeLock.isAcquiredInThisProcess()) の場合
                            ロックを解除します。
                    } キャッチ (例外 e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        （真）の間{
            実行者が終了した場合
                実行者.シャットダウン();
                システム。 out .println( "フラッシュセール終了後、残りの在庫は次のとおりです: " + stock.get());
            }
            TimeUnit.MILLISECONDS.sleep(100);
        }
    }
    プライベート静的CuratorFramework getZkClient() {
        文字列 zkServerAddress = "127.0.0.1:2181" ;
        ExponentialBackoffRetry 再試行ポリシー = 新しい ExponentialBackoffRetry(1000, 3, 5000);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerアドレス)
                .セッションタイムアウトMs(5000)
                .接続タイムアウトMs(5000)
                .retryPolicy(再試行ポリシー)
                。建てる（）;
        zkClient を起動します。
 zkClientを返します。
    }
 }

印刷結果は以下のとおりです。最初は、在庫の読み取りに対して 8 つの出力結果があります。現在の在庫は次のとおりです: 3. 次に、書き込みロックに戻り、在庫を順番に減算します。

在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
在庫を読み取り、現在の在庫: 3
購入完了、残り在庫数: 2
購入完了、残り在庫数: 1
購入完了、在庫残り: 0
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
在庫を読み取り、現在の在庫: 0
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足
フラッシュセールに参加、在庫不足

分散ロック選択

最も一般的に使用されるのは、Redis 分散ロックと Zookeeper 分散ロックです。パフォーマンスの面では、Redis の 1 秒あたりの TPS は簡単に数万に達します。大規模で同時実行性の高いシナリオでは、推奨される技術的ソリューションとして Redis 分散ロックを使用することをお勧めします。同時実行要件が特に高くない場合は、Zookeeper 分散処理を使用できます。