分散ロックには Redis と Zookeeper のどちらを使用すればよいでしょうか?

分散ロックを使用する理由は何ですか?

この問題について議論する前に、ビジネスシナリオを見てみましょう。

システム A は電子商取引システムであり、現在は 1 台のマシンに導入されています。システムにはユーザーが注文するためのインターフェースがありますが、注文する前に、ユーザーは在庫をチェックして、十分な在庫があるかどうかを確認する必要があります。

システムにはある程度の並行性があるため、商品の在庫は事前に redis に保存され、ユーザーが注文すると redis の在庫が更新されます。

この時点でのシステムアーキテクチャは次のとおりです。

しかし、これによって問題が発生します。ある瞬間に、Redis 内の特定の商品の在庫が 1 の場合、2 つのリクエストが同時に到着し、そのうちの 1 つは上図のステップ 3 まで実行され、データベースの在庫が 0 に更新されますが、ステップ 4 はまだ実行されていません。

他のリクエストはステップ 2 まで実行され、在庫がまだ 1 であることがわかったので、ステップ 3 に進みます。

結果、商品は 2 個売れましたが、在庫は 1 個しかありません。

これは明らかに間違っています!これは典型的な在庫の売られ過ぎの問題です。

この時点で、解決策を考えるのは簡単です。ロックを使用してステップ 2、3、4 をロックし、これらのステップが実行された後に別のスレッドがステップ 2 を実行できるようにします。

上図の通り、ステップ 2 を実行する際に、Java が提供する synchronized または ReentrantLock を使用してロックし、ステップ 4 が完了したらロックを解除します。

このように、ステップ 2、3、および 4 は「ロック」され、複数のスレッド間でのみ連続して実行できるようになります。

しかし、良い時代は長くは続かなかった。システム全体の同時実行性が急上昇し、1 台のマシンでは処理できなくなりました。次に、以下に示すようにマシンを追加する必要があります。

マシンを追加すると、システムは上記のようになります。すごいですね!

2 人のユーザーのリクエストが同時に到着したが、異なるマシンに届いたと仮定すると、これらの 2 つのリクエストを同時に実行できますか? それとも、在庫の過剰販売の問題が発生しますか?

なぜ？上図の 2 つの A システムは 2 つの異なる JVM で実行されるため、追加されるロックはそれぞれの JVM 内のスレッドに対してのみ有効であり、他の JVM 内のスレッドに対しては無効です。

したがって、ここでの問題は、Java が提供するネイティブロックメカニズムが、複数のマシンの展開シナリオでは機能しないことです。

これは、2 台のマシンによって追加されたロックが同じロックではない (2 つのロックが異なる JVM にある) ためです。

では、2 台のマシンに追加されたロックが同じロックであることを確認すれば、問題は解決するのではないでしょうか。

この時点で、分散ロックが登場する時が来ました。分散ロックの考え方は次のとおりです。

システム全体で、ロックを取得するためのグローバルかつ一意の「もの」が提供され、各システムはロックが必要なときにこの「もの」にロックを要求するため、異なるシステムによって取得されたロックは同じロックであるとみなすことができます。

この「もの」としては、Redis、Zookeeper、またはデータベースが考えられます。

テキストの説明はあまり直感的ではありませんので、次の図を見てみましょう。

上記の分析から、在庫過剰のシナリオでは、分散展開システムで Java のネイティブロックメカニズムを使用するとスレッドの安全性が保証されないため、分散ロックソリューションを使用する必要があることがわかります。

では、分散ロックをどのように実装するのでしょうか?読み続けてください！

Redis ベースの分散ロックの実装

上記の分析は、分散ロックが使用される理由を説明しています。ここでは、分散ロックが実装された場合にどのように対処するかについて詳しく説明します。拡張: Redisson は分散ロックをどのように実装しますか?

最も一般的な解決策は、分散ロックにRedisを使用することです。

分散ロックに Redis を使用するアイデアは、おおよそ次のようになります。Redis にロックが追加されたことを示す値を設定し、ロックが解除されたらキーを削除します。

具体的なコードは次のとおりです。

 // ロックを取得する 
 // NX はキーが存在しない場合は成功を意味し、キーが存在する場合は false を意味します。PX は有効期限を指定できます。  
 anyLockの一意の値NX PX 30000を設定する 
 // ロックを解除する: Lua スクリプトを実行する 
 // ロックの解放には2つの命令が必要であり、これらはアトミックではない 
 // Redis の Lua スクリプト サポート機能が必要です。 RedisはLuaスクリプトをアトミックに実行する 
 redis.call("get",KEYS[1]) == ARGV[1]の場合 
 redis.call("del",KEYS[1]) を返す
 それ以外 
 0を返す 
終わり

このアプローチにはいくつかの重要なポイントがあります。

必ずSETキー値NX PXミリ秒コマンドを使用してください。

そうでない場合は、まず値を設定し、次に有効期限を設定します。これはアトミック操作ではないため、有効期限を設定する前にクラッシュが発生し、デッドロックが発生する可能性があります (キーは永久に存在します)。

値は一意である必要があります

これは、ロック解除時にキーを削除する前に、値がロックされた値と一致していることを確認するためです。

これにより、A がロックを取得し、有効期限が 30 秒の場合、35 秒後にロックが自動的に解除されるという状況を回避できます。 A はロックを解除しようとしますが、この時点で B がロックを取得している可能性があります。クライアント A は B のロックを削除できません。

クライアントが分散ロックを実装する方法を考慮するだけでなく、Redis の展開も考慮する必要があります。

Redis には 3 つのデプロイメント方法があります。

スタンドアロンモード
マスタースレーブ + センチネル選挙モード
Redis クラスターモード

分散ロックに Redis を使用することの欠点は、単一マシンのデプロイメントモードを採用した場合、Redis が失敗する限り、単一ポイントの問題が発生することです。ロックしても機能しません。

マスタースレーブモードでは、ロック中に 1 つのノードのみがロックされます。センチネルによって高可用性が実現されていても、マスターノードに障害が発生してマスターとスレーブの切り替えが発生すると、ロックが失われる可能性があります。

上記の考慮に基づいて、redis の作成者もこの問題を考慮しました。彼は次のような意味を持つ RedLock アルゴリズムを提案しました。

Redis のデプロイメントモードが Redis Cluster であり、マスターノードが合計 5 つあると仮定します。以下の手順に従ってロックを取得します。

現在のタイムスタンプをミリ秒単位で取得する
各マスターノードに順番にロックを作成し、有効期限を短く設定します（通常は数十ミリ秒）。
多数のノードでロックを確立するようにしてください。たとえば、5つのノードの場合は3つのノード（n/2 +1）が必要です。
クライアントはロックを確立するのにかかる時間を計算します。ロックを確立するのにかかる時間がタイムアウトより短い場合、ロックは正常に確立されます。
ロックを確立できなかった場合は、順番にロックを削除します
他の誰かが分散ロックを確立している限り、ロックを取得するためにポーリングを続ける必要があります。

ただし、このアルゴリズムはまだ議論の余地があり、多くの問題がある可能性があり、ロックプロセスが正しいという保証はありません。

別の方法：レディソン

さらに、RedisクライアントのネイティブAPIに基づいてRedis分散ロックを実装するには、オープンソースフレームワークRedissionを使用することもできます。

Redisson は、分散ロックのサポートも提供するエンタープライズレベルのオープンソース Redis クライアントです。私も皆さんにこれを使うことを強くお勧めします。なぜでしょうか?

上で述べたことを思い出してください。 Redis を通じて値を設定する独自のコードを記述する場合は、次のコマンドを通じて設定します。

anyLockの一意の値NX PX 30000を設定する

ここで設定されるタイムアウトは 30 秒です。 30 秒以内にビジネスロジックを完了しないと、キーの有効期限が切れ、他のスレッドがロックを取得する可能性があります。

この場合、最初のスレッドがビジネスロジックの実行を完了する前に 2 番目のスレッドが実行される場合、スレッドの安全性の問題が発生します。なので、この有効期限も別途管理する必要があり、面倒ですね〜

redisson がどのように実装されているか見てみましょう。まずはredissionを使う喜びを感じてください:

設定config = new Config();  
 config.useClusterServers()  
 .addNodeAddress("redis://192.168.31.101:7001")  
 .addNodeAddress("redis://192.168.31.101:7002")  
 .addNodeAddress("redis://192.168.31.101:7003")  
 .addNodeAddress("redis://192.168.31.102:7001")  
 .addNodeAddress("redis://192.168.31.102:7002")  
 .addNodeAddress("redis://192.168.31.102:7003");  
 RedissonClient redisson = Redisson .create(config);  
 RLockロック= redisson .getLock("anyLock");  
ロック。ロック();  
ロックを解除します。

とても簡単です。分散ロックを完了するには、API のロック機能とロック解除機能を使用するだけです。それは多くの詳細を考慮するのに役立ちます:

すべてのredissonコマンドはLuaスクリプトを通じて実行され、redisはLuaスクリプトのアトミック実行をサポートしています。
Redisson はキーのデフォルトの有効期限を 30 秒に設定します。クライアントが 30 秒以上ロックを保持するとどうなりますか?

redisson にはウォッチドッグ (監視犬) という概念があります。ロックを取得した後、10 秒ごとにキーのタイムアウトを 30 秒に設定できるようになります。

この方法では、ロックが常に保持されていても、キーの有効期限が切れたり、他のスレッドがロックを取得したりといった問題は発生しません。

Redisson の「ウォッチドッグ」ロジックにより、デッドロックが発生しないことが保証されます。

(マシンがクラッシュするとウォッチドッグはなくなります。このとき、キーの有効期限は延長されません。30秒後に自動的に期限切れになり、他のスレッドがロックを取得できるようになります)

以下にその実装コードの一部を示します。

 // ロックロジック 
プライベート< T > RFuture < Long > tryAcquireAsync(long リースタイム、TimeUnit ユニット、final long スレッド ID) {  
    リースタイムが -1 の場合 
        tryLockInnerAsync(leaseTime、ユニット、スレッドId、RedisCommands.EVAL_LONG) を返します。
    }  
    // lua スクリプトを呼び出して、いくつかのキーと有効期限を設定します 
    RFuture <長い>   ttlRemainingFuture = tryLockInnerAsync (commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); 
     ttlRemainingFuture.addListener(新しいFutureListener < Long > () {  
        @オーバーライド 
        パブリック void operationComplete(Future < Long > future) は例外をスローします {  
            （！future.isSuccess()）の場合{  
                戻る;  
            }  
            Long ttlRemaining = future .getNow();  
            // ロックを取得しました 
            残り時間== null の場合 
                // ウォッチドッグロジック 
                スケジュール有効期限更新(スレッドID)  
            }  
        }  
    });  
    ttlRemainingFuture を返します。  
 }  
 < T > RFuture < T > tryLockInnerAsync(long リースタイム、TimeUnit ユニット、long スレッド ID、RedisStrictCommand < T >コマンド) {  
内部ロックリース時間= unit .toMillis(リース時間);  
    commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, コマンド, を返します。  
              「(redis.call('exists', KEYS[1]) == 0) の場合」 +  
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +  
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +  
                  "nil を返す; " +  
              「終了;」+  
              「(redis.call('hexists', KEYS[1], ARGV[2]) == 1) の場合」 +  
                  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +  
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +  
                  "nil を返す; " +  
              「終了;」+  
              "redis.call('pttl', KEYS[1]); を返します",  
                コレクション。 <オブジェクト> singletonList(getName())、internalLockLeaseTime、getLockName(threadId));
 }  
 // ウォッチドッグは最終的にこれを呼び出します 
プライベート void スケジュール更新 (最終ロング スレッド ID) {  
    有効期限更新マップにキーが含まれている場合(getEntryName()) {  
        戻る;  
    }  
    // このタスクは10秒の遅延で実行されます 
    タイムアウトタスク= commandExecutor .getConnectionManager().newTimeout(new TimerTask() {  
        @オーバーライド 
        パブリック void run(Timeout timeout) 例外をスローします {  
            // この操作により、キーの有効期限が 30 秒にリセットされます 
            RFuture <ブール値>  将来= renewExpirationAsync (スレッドID);  
            future.addListener(新しいFutureListener <Boolean> () {  
                @オーバーライド 
                パブリック void operationComplete(Future < Boolean > future) は例外をスローします {  
                    有効期限更新マップを削除します。  
                    （！future.isSuccess()）の場合{  
                        log.error("ロック " + getName() + " の有効期限を更新できません", future.cause());  
                        戻る;  
                    }  
                    （future.getNow()）の場合{  
                        // スケジュールを再設定する 
                        // このメソッドを再帰的に呼び出すと、有効期限が無限に延長されます 
                        スケジュール有効期限更新(スレッドID)  
                    }  
                }  
            });  
        }  
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);  
    if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {  
        タスクをキャンセルします。  
    }  
 }

さらに、redisson は redlock アルゴリズムのサポートも提供します。

使い方も非常に簡単です:

 RedissonClient redisson = Redisson .create(config);  
 RLock lock1 = redisson .getFairLock("lock1");  
 RLock lock2 = redisson .getFairLock("lock2");  
 RLock lock3 = redisson .getFairLock("lock3");  
 RedissonRedLockマルチロック=新しいRedissonRedLock(lock1、lock2、lock3);  
マルチロック。ロック();  
マルチロック。ロック解除();

まとめ：

このセクションでは、Redisを分散ロックとして使用する具体的な実装計画を分析します。

そしてその限界のいくつか

その後、Redis クライアントフレームワーク redisson が導入されました。

これを使用することをお勧めします。

自分でコードを書くよりも、細かいことはあまり気にしなくて済みます。

Zookeeper に基づく分散ロックの実装

一般的な分散ロック実装ソリューションとしては、Redis を使用するほか、Zookeeper を使用して分散ロックを実装することもできます。

Zookeeper が実装する分散ロックの仕組み (以下、ZK と呼びます) を紹介する前に、ZK とは何かを簡単に紹介しましょう。

Zookeeper は、構成管理、分散コラボレーション、命名機能を提供する集中型サービスです。

zk モデルは次のとおりです: zk には、ファイルシステムと同様に、znode と呼ばれる一連のノードが含まれており、各 znode はディレクトリを表し、znode にはいくつかの特性があります。

順序付けられたノード: 現在親ノード /lock が存在する場合、この親ノードの下に子ノードを作成できます。

Zookeeper はオプションの順序付け機能を提供します。たとえば、子ノード「/lock/node-」を作成し、順序を指定できます。その後、Zookeeper は子ノードを生成するときに、現在の子ノードの数に基づいて整数のシーケンス番号を自動的に追加します。

つまり、最初に作成された子ノードの場合、生成される子ノードは /lock/node-0000000000、次のノードは /lock/node-0000000001 というようになります。

一時ノード: クライアントは一時ノードを作成できます。セッションが終了するかタイムアウトすると、Zookeeper は自動的にノードを削除します。
イベント監視: データを読み取るときに、ノードのイベント監視を設定することもできます。ノードのデータまたは構造が変更されると、Zookeeper はクライアントに通知します。現在、Zookeeper には次の 4 つのイベントがあります。

ノードの作成
ノードの削除
ノードデータの変更
サブノードの変更

ZK の上記の特性に基づいて、ZK を使用して分散ロックを実装するためのソリューションを簡単に考え出すことができます。

ZK の一時ノードと順序付きノードを使用して、各スレッドは /lock/ ディレクトリなどの ZK に一時的な順序付きノードを作成することによってロックを取得します。
ノードが正常に作成された後、/lock ディレクトリ内のすべての一時ノードを取得し、現在のスレッドによって作成されたノードがすべてのノードの中で最小のシーケンス番号を持つノードであるかどうかを判断します。
現在のスレッドによって作成されたノードが、すべてのノードの中で最小のシーケンス番号を持つノードである場合、ロックが正常に取得されたとみなされます。
現在のスレッドによって作成されたノードが、すべてのノードの中で最小のシーケンス番号を持つノードでない場合は、前のノードシーケンス番号を持つノードにイベントリスナーが追加されます。
たとえば、現在のスレッドで取得されたノード番号が /lock/003 で、すべてのノードのリストが [/lock/001、/lock/002、/lock/003] の場合、/lock/002 ノードにイベントリスナーが追加されます。

ロックが解除されると、次のシーケンス番号を持つノードが起動され、ステップ 3 が再実行され、自身のノードシーケンス番号が最小であるかどうかが判断されます。

例えば、/lock/001 が解放され、/lock/002 が時間を検出すると、ノードセットは [/lock/002,/lock/003] となるため、シーケンス番号が最も小さいノードである /lock/002 がロックを取得します。

全体のプロセスは次のとおりです。

これが具体的な実装アイデアです。コードの書き方については、比較的複雑なのでここでは掲載しません。

キュレーターの紹介

Curator は Zookeeper のオープンソースクライアントであり、分散ロックの実装も提供します。

使い方も比較的簡単です。

 InterProcessMutex interProcessMutex =新しいInterProcessMutex(client,"/anyLock");  
インタープロセスミューテックスを取得します。  
プロセスミューテックスを解放します。

分散ロックを実装するためのコアソースコードは次のとおりです。

 private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) は例外をスローします 
 {  
    ブール値haveTheLock = false ;  
    ブール型doDelete = false ;  
    試す {  
        （取り消し可能なget()！= null）の場合{  
            client.getData().usingWatcher(revocableWatcher).forPath(ourPath);  
        }  
        ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) の間 {  
            // 現在のすべてのノードのソートされたセットを取得します 
            リスト<文字列>         子供= getSortedChildren ();  
            // 現在のノードの名前を取得します 
            文字列sequenceNodeName = ourPath .substring(basePath.length() + 1); // スラッシュを含めるには +1  
            // 現在のノードが最小のノードであるかどうかを判定する 
            PredicateResults predicateResults = driver .getsTheLock(client, children, sequenceNodeName, maxLeases);  
            if (述語結果.getsTheLock()) {  
                // ロックを取得する 
ロックを有効にする= true ;  
            } それ以外 {  
                // ロックが取得されていない場合は、現在のノードの前のノードのリスナーを登録します 
                文字列previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();  
                同期(これ){ 
                     Stat stat = client .checkExists().usingWatcher(watcher).forPath(previousSequencePath);  
                    （統計値が null の場合）  
                        if ( millisToWait != null ) {  
                            millisToWait - = (System.currentTimeMillis() - startMillis);  
 startMillis =システム.currentTimeMillis();  
                            （ミリ秒待機時間< = 0）の場合{  
削除する= true ; // タイムアウト - ノードを削除します 
                                壊す;  
                            } 
                             待機(ミリ秒単位の待機)。  
                        }それ以外{  
                            待って（）;  
                        }  
                    }  
                }  
                // そうでない場合は削除されている可能性があります（つまり、ロックが解除されています）。再度取得を試みる 
            }  
        } 
     }  
    キャッチ(例外e){  
削除する= true ;  
        eを投げる;  
    } ついに{  
        if (doDelete) {  
            パスを削除します。  
        }  
    }  
    haveTheLock を返します。  
 }

実際、キュレーターの分散ロック実装の基本原理は、上記の分析と似ています。ここでは、図を使ってその原理を詳しく説明します。

まとめ：

このセクションでは、分散ロックを実装するための zookeeperr のソリューションと zk のオープンソースクライアントの基本的な使用方法を紹介し、その実装原則を簡単に紹介します。関連資料として、「例を使って ZooKeeper で分散ロックを実装してみましょう!」を参照してください。

2つのソリューションの長所と短所の比較

このセクションでは、2 つの分散ロック実装について学習した後、redis と zk の実装の利点と欠点について説明します。

Redis 分散ロックの場合、次のような欠点があります。

ロックを取得する方法は単純かつ粗雑です。ロックの取得に失敗した場合は、ロックの取得を試行し続けるため、パフォーマンスが低下します。
さらに、Redis の設計により、データの一貫性が強くないことが判断され、極端な場合には問題が発生する可能性があります。ロックモデルは十分に堅牢ではない
レッドロックアルゴリズムが使用されたとしても、一部の複雑なシナリオでは 100% 問題がないとは保証されません。 redlock の詳細については、「分散ロックの実行方法」を参照してください。
Redis の分散ロックでは、実際には常にロックの取得を試行する必要があり、パフォーマンスが低下します。

しかし一方で、Redisを使用して分散ロックを実装することは多くの企業で非常に一般的であり、ほとんどの場合、いわゆる「非常に複雑なシナリオ」に遭遇することはありません。

したがって、分散ロックとして Redis を使用することも良い解決策です。最も重要な点は、Redis はパフォーマンスが高く、高並行性のロック取得および解放操作をサポートできることです。

zk 分散ロックの場合: