同時実行シナリオにおけるべき等性の問題 - 分散ロックの詳細な説明

前書き: この記事で説明するべき等性の問題はすべて、並行シナリオにおけるべき等性の問題です。つまり、システムは元々べき等設計でしたが、同時実行シナリオでは失敗しました。

1. 要約

この記事では、DingTalk の実在人物認証シナリオにおけるデータ重複の問題から始め、問題の原因を分析し、同時実行性がべき等性の障害を引き起こすことを突き止め、べき等性の概念を紹介します。

並行シナリオにおけるべき等性の問題に対応して、べき等性を実現するための実行可能な方法論が提案されています。アドレス帳に人を追加するというビジネスシナリオと組み合わせて、データベースの冪等性の問題を簡単に分析し、分散ロックを使用して冪等性を実現する方法について詳細に説明します。

この論文では、単一点障害、ネットワークタイムアウト、他者のロックの不正な解放、ロックの早期解放、分散ロックの単一点障害など、分散シナリオにおけるロックに存在する問題を分析し、対応するソリューションを提案し、対応するソリューションの具体的な実装を紹介します。

2番目の質問

DingTalkの実名認証事業では、データ重複の問題がある。

1 問題現象

通常の状況では、データベース内に成功した実在の人物の認証レコードは 1 つだけ存在する必要がありますが、実際には特定のユーザーには複数のレコードが存在する場合があります。

2 原因

並行性は非べき等性につながります。

まず、べき等性の概念を確認しましょう。

べき等性は数学とコンピュータサイエンスの概念であり、抽象代数でよく見られます。

プログラミングにおいて、べき等な操作とは、何度実行しても、その効果が 1 回実行した場合と同じになる操作のことです。

--百度百科事典より

実在人物による認証はビジネスにおいてべき等的な設計になっており、その一般的なプロセスは次のとおりです。

1) ユーザーが本人認証を選択すると、サーバー上でレコードが初期化されます。

2) ユーザーは、DingTalk モバイル端末の指示に従って顔照合を完了します。

3) 比較が完了したら、サーバーにアクセスしてデータベースの状態を変更します。

ステップ 3 では、データベースのステータスを変更する前に、データベースが初期化されているかどうか、実際の人物が認証されているかどうか、および Zhike が認証成功を返して冪等性を保証するかどうかを確認します。リクエストが初めてサーバーにアクセスしてデータベースの状態を変更しようとする場合にのみ、べき等性判定条件が満たされ、データベースの状態が変更されます。残りのリクエストは、データベースの状態に影響を与えることなく直接返されます。サーバーへの複数のアクセスを要求した結果は、サーバーへの最初のアクセスを要求した結果と同じになります。したがって、実在の人物による認証が成功しているという前提では、データベースには成功した認証レコードが 1 つだけ存在するはずです。

しかし、実際には、同じリクエストによってデータベースの状態が複数回変更され、システムが期待どおりに冪等性を達成していないことがわかりました。その理由は、同時アクセス要求により、最初の要求がサーバーの状態の変更を完了する前に、他の同時要求と最初の要求が冪等性判定を通過し、データベースの状態を複数回変更してしまうためです。

同時実行により、元のべき等設計が失敗します。

並行性は非べき等性につながります。

3つの解決策

同時実行シナリオで冪等性の問題を解決する鍵は、一意の制約を見つけ、一意性チェックを実行し、同じデータを 1 回保存し、同じ要求操作を 1 回実行することです。

サーバーへのアクセス要求により、次のやり取りが発生する可能性があります。

データベースのステータスの変更などのデータソースを操作します。
ダウンストリームサービスの呼び出しやメッセージの送信など、他のビジネスシステムと対話します。

リクエストには、1 つのインタラクションのみ、または複数のインタラクションを含めることができます。たとえば、リクエストによってデータベースのステータスを 1 回だけ変更したり、データベースのステータスが変更された後にデータベースのステータスが正常に変更されたことを示すメッセージを送信したりできます。

したがって、次のような結論を導き出すことができます。並行シナリオでは、システムが依存するコンポーネントがべき等であれば、システムは自然にべき等です。

データベースを例にとると、データに対するリクエストの影響が新しいデータの追加である場合、一意のインデックスはべき等性の問題の解決策になります。データベースは、同じデータが繰り返し保存されないように一意性チェックを実行するのに役立ちます。

DingTalk アドレス帳に人を追加すると、データベースの一意のインデックスを使用してべき等性の問題が解決されます。 DingTalk アドレス帳に人を追加する場合を例にとると、データベースにデータを書き込む前に、まずそのデータがデータベース内に既に存在するかどうかを判断します。そうでない場合、追加要求によって最終的にデータベースの従業員テーブルにデータが挿入されます。アドレス帳に人を追加するための同一の同時要求が多数発生すると、システムのべき等設計が失敗する可能性があります。人物を追加するリクエストでは、(組織 ID、従業員番号) によってリクエストを一意にマークできます。データベースには、（組織ID、従業員番号）の一意のインデックスも存在します。したがって、複数の同一のユーザー追加要求によってデータベースのステータスが 1 回だけ変更され、レコードが追加されることを保証できます。

依存コンポーネントが自然にべき等であれば、問題は単純になりますが、実際の状況はより複雑になることがよくあります。並行シナリオでは、システムが依存するコンポーネントがべき等性を持たない場合、べき等性を実現するために追加の手段を使用する必要があります。

一般的に使用される方法は、分散ロックを使用することです。分散ロックを実装する方法は多数ありますが、最も一般的に使用されるのはキャッシュされた分散ロックです。

4つの分散ロック

「Java 分散ロックとは何ですか?」には、次のような一節があります。

コンピュータサイエンスでは、ロックは、異なるスレッドが同じリソースを操作するのを防ぐためのマルチスレッド環境のメカニズムです。ロックを使用すると、リソースは特定のスレッドによるアクセスに対して「ロック」され、リソースが解放された後にのみ別のスレッドからアクセスできるようになります。ロックにはいくつかの利点があります。2 つのスレッドが同じ作業を行うことを防止し、2 つのスレッドが同じリソースを同時に使用しようとしたときにエラーやデータの破損を防ぎます。

Java の分散ロックは、同じマシン上で実行されている複数のスレッドだけでなく、分散システム内の異なるマシン上のクライアントで実行されているスレッドでも機能するロックです。これらの個別のマシン上のスレッドは、他のマシンによってロックされているリソースにいずれのスレッドもアクセスしないように、通信して調整する必要があります。

これらの段落では、ロックの本質は共有リソースへの相互排他的アクセスであり、分散ロックは分散システムにおける共有リソースへの相互排他的アクセスの問題を解決することが説明されています。

java.util.concurrent.locks パッケージは、フェアロック/アンフェアロック、ブロッキングロック/非ブロッキングロック、読み取り/書き込みロック、再入可能ロックなど、豊富なロック実装を提供します。

分散ロックをどのように実装するのでしょうか?

解決策1

分散システムには、2 つの一般的な問題があります。

1) 単一点障害問題、つまり、ロックを保持しているアプリケーションで単一点障害が発生すると、ロックは長時間無効になります。

2) ネットワークタイムアウトの問題。つまり、クライアントネットワークがタイムアウトしてもロックが実際に成功した場合、再度正しくロックすることができない。

ロックを取得します。

問題 1 を解決するための簡単な解決策は、有効期限 (リース時間) を導入することです。保持されるロックには時間制限があります。アプリケーションで単一点障害が発生した場合、そのアプリケーションが保持しているロックは自動的に解除されます。

問題 2 を解決するための簡単な解決策は、再入可能性をサポートすることです。ロックを取得するクライアントごとに一意の識別子 (通常は UUID) を構成します。ロックが正常にロックされると、ロックにはクライアントの識別子が保持されます。ロックが実際に成功し、クライアントがタイムアウト後に再試行すると、ロックがすでにクライアントによって保持されていると判断し、成功を返します。

要約すると、リースベースの分散ロックソリューションを提案しました。パフォーマンス上の理由から、キャッシュはロックの記憶媒体として使用され、共有リソースへの相互排他アクセスの問題を解決するために MVCC (マルチバージョン同時実行制御) メカニズムが使用されます。具体的な実装は付録のコードで確認できます。

分散ロックの一般的な使用法は次のとおりです。

分散ロックファクトリーを初期化する
ファクトリを使用して分散ロックインスタンスを生成する
この分散インスタンスを使用して操作をロックおよびロック解除します

@テスト
パブリックボイドテストトライロック（）{ 
 
    // ファクトリーを初期化する
    MdbDistributeLockFactory を新規 MdbDistributeLockFactory() に追加します。
    mdbDistributeLockFactory.setNamespace(603);
    mdbDistributeLockFactory.setMtairManager(新しい MultiClusterTairManager()); 
 
    //ロックを取得する
    DistributeLock ロック = mdbDistributeLockFactory.getLock( "TestLock" ); 
 
    //ロックとロック解除の操作
    ブール型ロック = lock.tryLock();
    （！ロックされている場合）{
戻る;
    }
    試す {
        //何かをする
    ついに
        ロックを解除します。
    }
 }

このソリューションはシンプルで使いやすいですが、問題点も明らかです。例えば、ロックを解除する際には、キャッシュ内のキーが単純に無効化されるため、誤って他人が保持しているロックを解除してしまうという問題があります。幸いなことに、ロックのリース期間が十分に長く設定されている限り、この問題が発生する可能性は十分に低くなります。

この問題を説明するために、Martin Kleppmann 氏の記事「分散ロックの実行方法」の図を使用します。

ロックを保持しているクライアント 1 がロックを解放する前にロックの有効期限が切れ、クライアント 2 がロックを取得する状況を想像してください。このとき、ロックはクライアント 2 によって保持されますが、クライアント 1 が誤ってロックを解除する可能性があります。より良い解決策は、ロックごとに ID を設定することです。ロックを解除するときは、1)まずロックが自分のものかどうかを確認し、2)自分のものであればロックを解除します。実装上の制限により、手順 1 と 2 はアトミック操作ではありません。手順 1 と 2 の間でロックの有効期限が切れ、他のクライアントによってロックが取得された場合、他のクライアントのロックが誤って解除されます。

解決策2

Redis の Lua スクリプトを使用すると、他のユーザーが保持しているロックを誤って解放してしまう問題を完全に解決できます。「Redis を使用した分散ロック」の記事の「単一インスタンスによる正しい実装」セクションでは、分散ロックの実装方法という必要な答えが得られます。

ロックを取得したい場合は、次のメソッドを実行します。

リソース名をランダム値に設定NX PX 30000

ロックを解除したいときは、次のLuaスクリプトを実行します。

 redis.call( "get" ,KEYS[1]) == ARGV[1]の場合 
 redis.call( "del" ,KEYS[1])を返す
それ以外 
 0を返す
終わり

オプション3

ソリューション 1 とソリューション 2 の議論の中で、ロックの自動解除という 1 つの問題が繰り返し言及されました。

これは諸刃の剣です。

1) 一方で、ロックを保持しているクライアントの単一障害点の問題を解決します。

2) 一方、ロックが早期に解除されると、ロックは誤って保持されます。

この時点で、Watch Dog の自動更新メカニズムを導入できます。 Redisson がこれをどのように実装するかについては、以下を参照してください。

ロックが正常に取得されると、Redisson は renewExpiration() メソッドを呼び出して Watch Dog スレッドを開始し、ロックを自動的に更新します。 1/3 ごとに 1 回更新します。成功した場合は、次の更新に進みます。失敗した場合は、更新操作をキャンセルします。

Redissonがどのようにリニューアルされたのか見てみましょう。 renewExpiration() メソッドの 17 行目 renewExpirationAsync() は、ロックの更新を実行するための重要な操作です。メソッドに入ると、Redisson も Lua スクリプトを使用してロックを更新していることがわかります。1) ロックが存在するかどうかを判断します。 2) 存在する場合は有効期限をリセットします。

プライベートvoid renewExpiration() {
    有効期限エントリ ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    ee == nullの場合{
戻る;
    } 
 
    タイムアウトタスク = commandExecutor.getConnectionManager().newTimeout(timeout -> {
        有効期限エントリ ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        ent == nullの場合{
戻る;
        }
        ロングスレッドId = ent.getFirstThreadId();
        スレッドIDがnullの場合
戻る;
        } 
 
        RFuture<ブール値> future = renewExpirationAsync(threadId);
        future.onComplete((res, e) -> {
            e != null の場合
                log.error( "ロック " + getRawName() + " の有効期限を更新できません" , e);
                EXPIRATION_RENEWAL_MAP.remove(getEntryName());
戻る;
            } 
 
            もし（res）{
                // スケジュールを再設定する
                有効期限を更新します。
            }それ以外{
                有効期限更新をキャンセルします( null );
            }
        });
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); 
 
    タスクのタイムアウトを設定します。
 }

保護された RFuture<Boolean> renewExpirationAsync(long threadId) {
 evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,を返します。
 「(redis.call('hexists', KEYS[1], ARGV[2]) == 1) の場合」 +
 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
 "1を返す; " +
 「終了;」 +
 "0を返します。" 、
                          コレクション.singletonList(getRawName())、
                          内部LockLeaseTime、getLockName(threadId));
 }

オプション4

Redisson の自動更新メカニズムにより、ロックの自動解除について心配する必要がなくなりました。しかし、議論のこの時点では、まだ問題に直面しています。分散ロック自体は分散アプリケーションではないのです。 Redis サーバーが正常に動作しなくなると、分散ロック全体がサービスを提供できなくなります。

さらに、「Redis を使用した分散ロック」の記事で説明されている Redlock アルゴリズムとその実装を見てみましょう。

Redlock アルゴリズムは万能薬ではなく、その長所と短所については多くの議論があります。

分散ロックを実行する方法:

https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html

Redlock は安全ですか?

http://antirez.com/news/101

Martin Kleppmann と Antirez が Redlock について議論:

https://news.ycombinator.com/item

参考文献

Java 分散ロックとは何ですか?

https://redisson.org/glossary/java-distributed-lock.html

分散ロックとシンクロナイザー:

https://github.com/redisson/redisson/wiki/8.-distributed-locks-and-synchronizers

Redis による分散ロック:

https://redis.io/topics/distlock?spm=ata.21736010.0.0.31f77e3aFs96rz

付録

分散ロック

パブリッククラスMdbDistributeLockはDistributeLockを実装します{ 
 
    /**
     * 名前空間をロックする
     */
    プライベートな最終int名前空間; 
 
    /**
     * ロックに対応するキャッシュキー 
     */
    プライベート最終文字列 lockName; 
 
    /**
     * ロックの一意の識別子。putが成功したがタイムアウトが返された場合に再入可能性を保証します。
     */
    プライベート最終文字列 lockId; 
 
    /**
     * ロックを保持するかどうか。真: はい
     */
    プライベートブール値がロックされています。 
 
    /**
     * キャッシュインスタンス
     */
    プライベートファイナルTairManager tairManager; 
 
パブリックMdbDistributeLock(TairManager tairManager、 int名前空間、String lockCacheKey) { 
 
        this.tairManager = tairManager;
        this.namespace = 名前空間;
        this.lockName = lockCacheKey;
        this.lockId = UUID.randomUUID().toString();
    } 
 
    @オーバーライド
パブリックブール型tryLock() { 
 
        試す {
            //ロックステータスを取得する
            Result<DataEntry> getResult = null ;
            結果コード getResultCode = null ;
 ( int cnt = 0; cnt < DEFAULT_RETRY_TIMES; cnt++) {
                getResult = tairManager.get(名前空間、ロック名);
                getResultCode = getResult == null ? null : getResult.getRc();
                再試行が必要ない場合(getResultCode) {
                    壊す;
                }
            } 
 
            // 再入力、ロックはすでに保持されている、成功を返す
            if (ResultCode.SUCCESS.equals(getResultCode)
                && getResult.getValue() != null && lockId.equals(getResult.getValue().getValue())) {
                ロック = true ;
戻る 真実;
            } 
 
            //ロックを取得できないため、失敗を返します
            if (!ResultCode.DATANOTEXSITS.equals(getResultCode)) {
                log.error( "tryLock 失敗コード={} ロック={} トレースId={}" 、getResultCode、this、EagleEye.getTraceId());
戻る 間違い;
            } 
 
            //ロックを取得しようとする
            結果コード putResultCode = null ;
 ( int cnt = 0; cnt < DEFAULT_RETRY_TIMES; cnt++) {
                putResultCode = tairManager.put(名前空間、ロック名、ロックID、MDB_CACHE_VERSION、
                    デフォルト値:
                再試行が必要ない場合(putResultCode) {
                    壊す;
                }
            }
            if (!ResultCode.SUCCESS.equals(putResultCode)) {
                log.error( "tryLock 失敗コード={} ロック={} トレースId={}" 、getResultCode、this、EagleEye.getTraceId());
戻る 間違い;
            }
            ロック = true ;
戻る 真実; 
 
        } キャッチ (例外 e) {
            log.error( "DistributedLock.tryLock がロックに失敗しました = {}" 、 this、 e);
        }
戻る 間違い;
    } 
 
    @オーバーライド
パブリックボイドロック解除() { 
 
        （！ロックされている場合）{
戻る;
        }
        結果コード resultCode = tairManager.invalid(namespace, lockName);
        結果コードが成功の場合
            log.error( "DistributedLock.unlock 失敗 ロック={} resultCode={} traceId={}" , this, resultCode,
                EagleEye.getTraceId());
        }
        ロック = false ;
    } 
 
    /**
     * 再試行が必要かどうかを判断する
     *
     * @param resultCode キャッシュされた戻りコード
     * @戻る  true : 再試行しない
     */
    プライベートブール値 noNeedRetry(ResultCode 結果コード) {
結果コードがnull ではないこと、および !ResultCode.CONNERROR.equals(結果コード) および !ResultCode.TIMEOUT.equals(を返す
            結果コード) && !結果コード.UNKNOW.equals(結果コード);
    } 
 
 }

分散ロックファクトリー

パブリッククラスMdbDistributeLockFactoryはDistributeLockFactoryを実装します{ 
 
    /**
     * キャッシュ名前空間
     */
    @セッター
    プライベートint名前空間; 
 
    @セッター
    プライベート MultiClusterTairManager mtairManager; 
 
    @オーバーライド
パブリックDistributeLock getLock(String lockName) {
新しい MdbDistributeLock(mtairManager、名前空間、lockName)を返します。
    }
 }