長くて薄い10,000語の記事でRedisson分散ロックのソースコードを説明します

[[382196]]

序文

前回の記事ではRedisの分散ロックの原理と欠点について書きましたが、それだけでは不十分だと感じました。 Redisson フレームワークについては簡単に紹介しただけで、具体的な原則については説明していませんでした。旧正月が終わったら何もすることがないので、Redisson のソースコードを勉強してみるのもいいかもしれません。

気まぐれではありましたが、よく調べてみると、Redisson のソースコードを解釈する作業量は依然としてかなり多いことがわかりました。多数の Java 並行処理クラスが使用されており、Redis コンポーネントとのリモート呼び出しを実装するための通信ツールとして Netty が挙げられます。これらすべての知識ポイントを説明するのは現実的ではありません。この記事の焦点は主に Redisson 分散ロックの実装原理にあるため、ネットワーク通信と同時実行原理のコード解釈についてはあまり詳しく説明しません。至らぬ点がありましたらご容赦ください。

Redis のパブリッシュとサブスクライブ

前述したように、分散ロックのコア機能は実際にはロック、ロック解除、ロックタイムアウトの設定の 3 つです。これら 3 つの機能は、Redisson 分散ロックの原理に関する私たちの研究の方向性でもあります。

学習する前に、Redis のパブリッシュ/サブスクライブ機能に関する知識ポイントを理解する必要があります。

Redis のパブリッシュ/サブスクライブ (pub/sub) は、送信者 (pub) がメッセージを送信し、サブスクライバー (sub) がメッセージを受信するためのメッセージ通信モードです。パブリッシャーは指定されたチャネル（チャンネル）にメッセージを送信することができ、サブスクライバーはチャネルを購読することでメッセージを受信できるため、複数のクライアント間での通信効果が得られます。

サブスクリプションコマンドは SUBSCRIBE channel [channel ...] であり、1 つ以上のチャネルをサブスクライブできます。 PUBLISH コマンドを介して新しいメッセージがチャネルに送信されると、サブスクライバーは次のようにメッセージを受信します。

2 つのクライアントを開き、1 つはチャネル channel1 をサブスクライブし、もう 1 つが PUBLISH を通じてメッセージを送信すると、サブスクライブしたクライアントがそれを受信します。このモードでは、異なるクライアント間の通信を実現できます。

この通信モードの素晴らしい使用シナリオについては詳しく説明しません。オンラインで確認できます。私たちの主人公は依然としてレディソンです。ウォーミングアップが終わったら、メインコースの提供です。

Redisson ソースコード

Redisson を使用してロックする前に、まず RLock インスタンスオブジェクトを取得する必要があります。このオブジェクトを使用すると、lock メソッドと tryLock メソッドを呼び出してロック機能を完了できます。

設定 config = new Config();
 config.useSingleServer()
  .setPassword( "" )
  .setアドレス( "redis://127.0.0.1:6379" );
 RedissonClient redisson = Redisson.create (config);
 //RLock オブジェクト
RLock ロック = redisson.getLock( "myLock" );

対応するホストを構成した後、RLock オブジェクトを作成できます。 RLock は、特定の同期装置が実装する必要があるインターフェースです。 redisson.getLock()を呼び出すと、プログラムはデフォルトの同期エグゼキュータRedissonLockを初期化します。

ここでいくつかのパラメータが初期化されます。

commandExecutor : 非同期エグゼキューター。 Redisson のすべてのコマンドは...Executor を通じて実行されます。

id : 初期化時に UUID を使用して作成された一意の ID。

internalLockLeaseTime : ロックを取得するまでの待機時間。構成クラスで定義されているデフォルトの時間は 30 秒です。

同時に、図には getEntryName メソッドもマークされています。このメソッドは、「ID: ロック名」という文字列を返します。これは、対応するロックを保持している現在のスレッドの識別子を表します。これらのパラメータは、後続のソースコード分析で頻繁に表示されるため、記録しておく必要があります。

初期化について説明した後、ロックとロック解除のソースコードの学習を開始できます。

ロック

Redisson には、tryLock と lock という 2 つのロックメソッドがあります。使用方法の違いは、tryLock ではロックの有効期限のleaseTime と待機時間のwaitTime を設定できることです。コア処理ロジックは同様です。まずはtryLockから始めましょう。

ロックを試みる

コードが少し長いです。。。写真にするのは不便なので、そのまま貼り付けておきます。

 /**
 * @param waitTime ロックを待つ時間の長さ
 * @paramleasingTime ロック保持時間
 * @param 単位 時間単位
 * @戻る 
 * @throws 中断例外
 */
 public boolean tryLock(long waitTime, longleasingTime, TimeUnit unit) throws InterruptedException { // ロックの残り待機時間
        長い時間= unit.toMillis(waitTime);
        長い現在の= System.currentTimeMillis(); 
         
        最終的な長いスレッドId = Thread.currentThread().getId();
        // ロックを取得しようとします。ロックが取得されなかった場合は、ロックの残りタイムアウトが返されます。
        Long ttl = tryAcquire(leaseTime, Unit, threadId);
        // ttl がnullの場合、ロックを取得できることを示し、 trueを返します 
        ttl == null の場合
戻る 真実;
        } 
         
        // waitTimeがタイムアウトした場合は、ロックの適用が失敗したことを示すfalseを返します。
時間-= (System.currentTimeMillis() -現在の値);
        もし (時間<= 0 ) {
            取得に失敗しました(スレッドID);
戻る 間違い;
        } 
         
現在の= System.currentTimeMillis();
        // 分散ロックをサブスクライブし、ロックが解除されたときに通知します。ここでは、上で説明したパブリッシュ/サブスクライブ方式を使用します。
        最終的な RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // ブロックしてロックが解除されるのを待ちます。 await() がfalseを返す場合、待機がタイムアウトしたことを意味します。
        if (!await(subscribeFuture, time , TimeUnit.MILLISECONDS)) {
            subscribeFuture.cancel( false )の場合
                subscribeFuture.addListener(新しいFutureListener<RedissonLockEntry>() {
                    @オーバーライド
パブリックvoid operationComplete(Future<RedissonLockEntry> future) は例外をスローします {
                        subscribeFuture.isSuccess() の場合 {
                         // 待機時間が経過したので、サブスクリプションをキャンセルします
                            購読を解除します(subscribeFuture、スレッドID);
                        }
                    }
                });
            }
            取得に失敗しました(スレッドID);
戻る 間違い;
        } 
 
        試す {
時間-= (System.currentTimeMillis() -現在の値);
            もし (時間<= 0 ) {
                取得に失敗しました(スレッドID);
戻る 間違い;
            }
         // 無限ループに入り、tryAcquire を繰り返し呼び出してロックの取得を試みます。ロジックは上記のロック取得ロジックと同じです。
            （真）の間{
                長いcurrentTime = System.currentTimeMillis();
                ttl = tryAcquire(リース時間、ユニット、スレッドID);
                // ロックを取得しました
                ttl == null の場合
戻る 真実;
                } 
 
時間-= (System.currentTimeMillis() - currentTime);
                もし (時間<= 0 ) {
                    取得に失敗しました(スレッドID);
戻る 間違い;
                } 
 
                //メッセージを待機中
                現在の時刻 = System.currentTimeMillis();
                ttl >= 0 && ttl <時間 の場合{
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }それ以外{
                    getEntry(threadId).getLatch().tryAcquire( time , TimeUnit.MILLISECONDS);
                } 
 
時間-= (System.currentTimeMillis() - currentTime);
                もし (時間<= 0 ) {
                    取得に失敗しました(スレッドID);
戻る 間違い;
                }
            }
        ついに
            購読を解除します(subscribeFuture、スレッドID);
        }
 // get(tryLockAsync(waitTime,leasingTime,unit));を返します。
    }

コードはまだかなり長いですが、プロセスはたった 2 つのステップです。スレッドがロックを取得して正常に戻るか、または、ロックの取得に失敗し、待機時間が経過していない場合は、ロックを取得するためにループを継続し、ロックが解除されるかどうかを監視します。

ロックを取得するメソッドは tryAcquire です。渡されるパラメータは、ロック保持時間、時間単位、および現在のスレッドを表す ID です。コードに従ってコールスタックを表示すると、tryAcquireAsync というメソッドが呼び出されます。

プライベート Long tryAcquire(long リースタイム、TimeUnit ユニット、long スレッド ID) {
 get(tryAcquireAsync(leaseTime, unit, threadId)) を返します。
 } 
 
プライベート <T> RFuture<Long> tryAcquireAsync(long リースタイム、TimeUnit ユニット、final long スレッド ID) {
        // ロックの待機時間が設定されている場合は、tryLockInnerAsync メソッドを直接呼び出してロックを取得します
        リースタイムが -1 の場合
tryLockInnerAsync(leaseTime、ユニット、スレッドId、RedisCommands.EVAL_LONG)を返します。
        }
        // ロックの待機時間が設定されていない場合は、別のリスナーを追加します。つまり、lock.lock() を呼び出すロジックが実行されます。これについては後で説明します。
        RFuture<Long> 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を返します。
    }

続けて、tryLockInnerAsync メソッドのソースコードを見てみましょう。

 <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]);を返します。」 、
                コレクション。<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
 }
文字列 getLockName(long threadId) {
 id + ":" + threadIdを返します。
 }

基礎となる呼び出しスタックは次のとおりです。コマンドを直接操作し、Lua スクリプトに統合し、netty ツールクラスを呼び出して redis と通信し、ロックを取得する機能を実現します。

このスクリプトコマンドは非常に興味深いので、簡単に説明しましょう。

まず、exists キーコマンドを使用して、ロックが使用されているかどうかを判断します。そうでない場合は、hset コマンドを使用して書き込みます。キーはロックの名前、フィールドは「クライアントの一意の ID: スレッド ID」、値は 1 です。
ロックが占有されている場合、現在のスレッドによって占有されているかどうかを判断し、占有されている場合は値に 1 を追加します。
ロックは現在のスレッドによって占有されておらず、ロックの残りの有効期限が返されます。

コマンドのロジックは複雑ではありませんが、作者の設計は非常に思慮深いものであると言わざるを得ません。データの保存には、Redis のハッシュ構造が使用されます。現在のスレッドがすでにロックを保持していることが判明した場合、hincrby コマンドを使用して値に 1 を追加します。 value の値によって、ロックが解除されたときに unlock コマンドが呼び出される回数が決定され、それによってロックの再入効果が実現されます。

以下の図にコマンドの各ステップに対応するロジックをマークしましたので、ご覧ください。

コードを続行しましょう。上記のコマンドによれば、スレッドがロックを取得すると、tryLock メソッドは直接 true を返し、すべて正常になります。

取得できない場合は、ロックの残り有効期限が返されます。今回の目的は何ですか？ tryLock メソッドのデッドループに戻りましょう。

ここでは、waitTime とキーの残りの有効期限を比較し、2 つのうち小さい方の値を取得し、Java の Semaphore セマフォの tryAcquire メソッドを使用してスレッドをブロックします。

では、Semaphore セマフォを制御するのは誰で、いつリリースできるのでしょうか?ここで上記に戻る必要があります。上記で投稿した tryLock コードに次の段落があることを覚えておいてください。

現在の= System.currentTimeMillis();
 // 分散ロックをサブスクライブし、ロックが解除されたときに通知します
最終的な RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);

サブスクリプションロジックは明らかに subscribe メソッドにあります。メソッド呼び出しチェーンに従って、PublishSubscribe.java に入ります。

このコードの目的は、現在のスレッドの threadId を AsyncSemaphore に追加し、redis の公開およびサブスクリプション機能を通じて実装される redis リスナーを設定することです。

リスナーは Redis からメッセージを受信すると、現在のスレッドに関連する情報を取得します。ロックが解除されたというメッセージであれば、Semaphore を操作する（つまり release メソッドを呼び出す）ことで、ブロックされていた領域を直ちに解放します。

解放後もスレッドは実行を継続し、タイムアウトしたかどうかを判断します。タイムアウトになっていない場合は、次のループに入り、再度ロックを取得します。ロックを取得すると true を返します。そうでない場合は、プロセスが続行されます。

ここで説明すると、ループが発生する理由は、ロックが複数のクライアントによって同時に競合される可能性があるためです。ブロッキングが解除された直後はスレッドがロックを取得できない可能性がありますが、スレッドの待機時間は経過していません。このとき、ロックを取得するにはループを再度実行する必要があります。

これは tryLock がロックを取得するプロセス全体です。フローチャートを描くと次のようになります。

ロック

tryLock に加えて、ロックを取得するために lock を直接呼び出すこともよくあります。 lock のロック取得プロセスは基本的に tryLock と同じです。違いは、ロックではロック有効期限パラメータが手動で設定されないことです。このメソッドの呼び出しチェーンは、ロックを取得するために tryAcquire メソッドまで実行されます。違いは、ロジックのこの部分まで実行されることです。

このコードは次の 2 つのことを行います。

1. 有効期限を30秒に設定してロックを取得します

2. リスナーを起動します。ロックが取得されたことが判明した場合は、スケジュールされたタスクを開始して、ロックの有効期限を継続的に更新します。

有効期限を更新する方法は、scheduleExpirationRenewal です。ソースコードは次のとおりです:

プライベート void スケジュール更新 (最終ロング スレッド ID) {
 // expirationRenewalMap は、「現在のスレッド ID: キー名」としてマークされたタスクを格納する ConcurrentMap です。
        有効期限更新マップにキーが含まれている場合(getEntryName()) {
戻る;
        } 
 
        タイムアウトタスク = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @オーバーライド
パブリックvoid run(Timeout timeout) 例外をスローします {
                // ロックが存在するかどうかを検出する Lua スクリプト。存在する場合は、pexpireコマンドを使用して有効期限を更新します。
                RFuture<ブール値> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
 「(redis.call('hexists', KEYS[1], ARGV[2]) == 1) の場合」 +
 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
 "1を返す; " +
 「終了;」 +
 "0を返します。" 、
                          コレクション。<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); 
                 
                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(), task) != null ) {
            タスクをキャンセルします。
        }
    }

コードフローは比較的シンプルです。基本的には、時間指定のタスクを開始し、internalLockLeaseTime / 3 (この時間は 10 秒) ごとに、現在のスレッドによってロックがまだ保持されているかどうかを確認します。その場合は、有効期限 internalLockLeaseTime を 30 秒にリセットします。

これらのスケジュールされたタスクは、ConcurrentHashMap オブジェクトの expireRenewalMap に保存され、保存されるキーは「スレッド ID: キー名」です。対応する現在のスレッドキーが expirationRenewalMap に存在しないことが判明した場合、スケジュールされたタスクは実行されません。これはその後のロック解除においても重要な操作となります。

上記のコードは、Redisson のいわゆる「ウォッチドッグ」プログラムであり、手動でロックを解除する前に期限が切れないように、非同期スレッドを使用して定期的に検出して実行します。

残りのロジックは基本的に tryLock() と同じです。見ればわかりますよ。

ロック解除

ロックする方法があれば、ロックを解除する方法もあります。 Redisson 分散ロックのロックを解除するための上位レベルの呼び出しメソッドは unlock() であり、デフォルトではパラメーターは渡されません。

 @オーバーライド
パブリックボイドロック解除() {
     // ロックを解除するためのコマンド要求を開始する
        ブール opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));
        opStatus == nullの場合
            throw new IllegalMonitorStateException( "ロックのロックを解除しようとしましたが、ノード ID によって現在のスレッドによってロックされていません: "  
                    + id + " スレッドID: " + Thread.currentThread().getId());
        }
        （opStatus）の場合{
         // ロックを正常に解放し、 「ウォッチドッグ」タイマー スレッドをキャンセルします
            有効期限の更新をキャンセルします。
        }
    }

ロック解除関連のコマンド操作は、unlockInnerAsync メソッドで定義されます。

これは別の長い Lua スクリプトの文字列であり、前のロックスクリプトよりも少し複雑です。しかし、それは問題ではありません。簡単に整理してみましょう。コマンドのロジックはおおよそ次のようになります。

1. ロックが存在するかどうかを判断します。そうでない場合は、publish コマンドを使用してメッセージを公開し、ロックを解除します。加入者はそれを受け取った後、ロックを取得する次のステップに進むことができます。

2. ロックは存在しますが、現在のスレッドによって保持されていないため、nil が返されます。

3. 現在のスレッドがロックを保持します。 hincrby コマンドを使用して、ロックの再入回数を 1 減らします。次に、再入回数が 0 より大きいかどうかを判断します。大きい場合は、ロックの有効期限を更新して 0 を返します。それ以外の場合は、ロックを削除し、ロックを解除するメッセージを公開して 1 を返します。

スレッドがロックを完全に解放すると、cancelExpirationRenewal() メソッドを呼び出してウォッチドッグの拡張スレッドをキャンセルします。

 void 有効期限更新をキャンセルする() {
 // 有効期限更新マップは対応するキーを削除し、現在のスレッドに対応する「ウォッチドッグ」プログラムは実行されません。
    タイムアウトタスク = expireRenewalMap.remove(getEntryName());
    タスクがnullの場合
        タスクをキャンセルします。
    }
 }

これはロックを解除するプロセスです。どうですか？これは比較的シンプルで、ロックを追加するためのコードよりもはるかに読みやすいです。もちろん、シンプルですが、分散ロックのプロセス全体を理解していただくために、フローチャートを描くことに苦労しました (この時点で、トリプルクリックしていただく必要がありますね、笑)。

レッドロック

以上がRedisson分散ロックの原理の説明です。一般的には、Lua スクリプトを使用して基本的な set コマンドを統合し、ロック機能を実装するだけです。これは、多くの Redis 分散ロックツールの設計原則でもあります。さらに、Redisson はロック効果を実現するための「RedLock アルゴリズム」の使用もサポートしています。このツールクラスは RedissonRedLock です。

使い方も非常に簡単です。複数の Redisson ノードを作成します。これらの無関係なノードは完全な分散ロックを形成できます。

 RLock lock1 = Redisson.create ( config1).getLock(lockKey);
 RLock lock2 = Redisson.create (config2).getLock(lockKey);
 RLock lock3 = Redisson.create (config3).getLock(lockKey); 
 
ロック1、ロック2、ロック3を設定します。
試す {
   ロックを解除します。
ついに
   redLock.ロック解除();
 }

RedLock アルゴリズムの原理については詳しく説明しません。ご興味がございましたら、前回の記事を読んだり、オンラインで検索したりしてみてください。つまり、Redis インスタンスの単一障害点の問題をある程度効果的に防ぐことができますが、完全に信頼できるわけではありません。どちらの設計が使用されても、Redis 自体はロックの強力な一貫性を保証することはできません。

古いことわざにもあるように、ケーキを食べてケーキも残すことはできませんが、これはパフォーマンスとセキュリティにも当てはまります。 Redis の強力なパフォーマンスと使いやすさは、日常の分散ロックのニーズを満たすのに十分です。ビジネスシナリオでロックのセキュリティリスクを許容できない場合、最も確実な方法は、ビジネスレイヤーでべき等処理を実行することです。

要約する

この記事のソースコード分析を読んだ後、Redisson 分散ロックの設計について十分に理解できたと思います。もちろん、ソースコードについては説明していますが、主な焦点は依然として分散ロックの原則にあります。プロセスに関連しない一部のコードは、逐語的に解釈されません。興味があれば自分で読んでみてください。ソースコードの多くの箇所で、いくつかの基本的な並行処理ツールとネットワーク通信の有用性が示されています。彼らから学ぶことは非常にやりがいがあります。

最後に、Redisson のコメントが本当に少ないことにまだ不満を言いたいです。。。。。。

<<: クラウドコンピューティングの未来はどうなるのか

>>: マルチクラウド戦略が組織のクラウドへの移行を簡素化する方法