べき等コンシューマー パターン

同じメッセージを複数回処理すると、1 回の処理と同じ効果が得られるように、メッセージ コンシューマーを設計します。 少なくとも 1 回の配信を保証するメッセージング システムは、同じメッセージを複数回配信できます。 重複に対する回復性がなければ、メッセージを再処理すると、重複するレコードが作成されたり、顧客に二重請求されたり、その他の望ましくない影響が生じたりする可能性があります。

コンテキストと問題

分散アプリケーションは、通常、直接同期呼び出しではなく、メッセージ ブローカーを介して作業を交換します。 Azure Service Bus、Azure Event Hubs、Apache Kafka、RabbitMQ などのほとんどのブローカーは、少なくとも 1 回の配信を提供します。 この保証は、エラーが発生した場合でもメッセージがコンシューマーに到達することを保証しますが、ブローカーが同じメッセージを複数回配信できることを意味します。

重複は、いくつかのソースから発生します。

  • プロデューサーの再試行

    プロデューサーはメッセージを送信し、一時的なネットワーク 障害またはタイムアウトのために受信確認を受信せず、メッセージを再度送信します。 送信が初めて成功した場合でも、ブローカーは 2 つのコピーを保持するようになりました。

  • 受信確認が見つからない後に再配信する

    コンシューマーはメッセージを受信して処理しますが、コンシューマーがクラッシュしたり、ロックが期限切れになったり、受信確認が失われたりしたため、メッセージの受信確認に失敗します。 ブローカーは、メッセージが処理されなかったと想定し、再度配信します。

  • 処理途中のコンシューマー障害

    コンシューマーはデータベースの書き込みを完了しますが、メッセージを受信確認する前にクラッシュします。 別のインスタンスが再配信されたメッセージを受け取ると、書き込みが繰り返されます。

分散システム全体での 1 回限りの配信は、保証には現実的ではありません。 厳密に 1 回のセマンティクスを要求するブローカーでも、コンシューマーにメッセージを配信したり、ブローカーにデータを書き戻したりするなど、直接制御する操作のみが保証されます。 コンシューマーが他のシステムで実行する外部の副作用を保証することはできません。 永続的なソリューションは、重複する配信を排除するためではありません。 これは、コンシューマーが許容できるようにするためです。 重複を無視するコンシューマーと少なくとも 1 回の配信を組み合わせると、 実質的に 1 回だけ処理を実行できます。

ソリューション

処理済みメッセージの記録を保持し、既に確認済みのメッセージをスキップすることで、コンシューマーを冪等にします。 コンシューマーは、この決定を、再配信後も存続する安定した識別子にキーを設定し、永続的なストアをチェックして、その識別子が既に処理されているかどうかを判断し、メッセージを処理するか、重複として破棄します。

次の手順では、コア フローについて説明します。

  1. メッセージを読み取り、その重複除去キーを抽出します。
  2. そのキーに対応する重複排除ストアを確認します。
  3. キーが存在する場合は、メッセージを重複として扱います。 それを確認して停止し、必要に応じて、以前に記録された結果を返します。
  4. キーが存在しない場合は、メッセージを処理し、1 つのアトミック操作でキーを記録してから、メッセージを確認します。

安定した重複除去キーを選択する

キーは、再配信のたびに論理メッセージを一意かつ一貫して識別する必要があります。 プロデューサー割り当てメッセージ識別子または特定の論理操作を識別するビジネス レベルのべき等キーを使用します。複数のメッセージが伝達できる共有の関連付けコンテキストではありません。 Azure Service Busでは、メッセージとそのペイロードを一意に識別するため、MessageId プロパティがこの目的を果たします。 CorrelationIdをキーとして使用しないでください。これは、要求とその応答などの関連するメッセージをグループ化するためです。 CloudEvents 仕様に従うイベントの場合、 source 属性と id 属性の組み合わせによってイベントが一意に識別され、再配信間で安定した状態が維持されます。

再配信時にブローカーが再生成するトランスポートレベルの識別子や、配信試行に由来する値をキーとして使用しないでください。これらの値は重複メッセージごとに変化し、重複検出を妨げるためです。 また、受信タイムスタンプなどの揮発性フィールドからキーを派生しないようにします。

パブリッシュ/サブスクライブ設計の複数のサブスクライバーなど、複数の独立したコンシューマーが同じチャネルを処理する場合、各コンシューマーはメッセージの独自のコピーを正当に処理し、メッセージ処理の完了を個別に追跡する必要があります。 これらのコンシューマーが 1 つの重複除去ストアを共有する場合は、コンシューマー ID とメッセージ ID の複合レコードにキーを設定します。 メッセージ ID のみをキーとするストアでは、最初のコンシューマーが他のすべてのコンシューマーによる処理を抑止してしまいます。

処理されたキーを格納する場所を決定する

次の 2 つの一般的なオプションがあります。

  • 専用の重複除去テーブル。 コンシューマーは、処理されたキーごとに 1 行を保持する別のテーブル ( 受信トレイと呼ばれることもあります) を保持します。 この方法では、重複除去の問題がビジネス データから分離され、多くのメッセージの種類が 1 つのメカニズムを共有している場合に適しています。

  • ビジネス エンティティ自体。 コンシューマーは、メッセージが作成または更新するレコードにキーを格納します。 この方法では、別のテーブルを回避できますが、重複除去がビジネス データの形に結合されます。

マーカーと副作用をアトミックにコミットする

check-then-process フローにはエラー ウィンドウがあります。 コンシューマーがメッセージを処理し、別の手順でキーを記録した場合、2 つの操作の間でクラッシュすると、副作用は適用されますが、キーは記録されないため、次の配信でメッセージが再処理されます。

重複排除マーカーとビジネス上の副作用を同じトランザクションに書き込むことで、この障害発生の余地に対処します。 両方が同時にコミットされるか、まったくコミットされない場合、再配信では、コミット済みのマーカーを見つけて処理をスキップするか、トランザクションがロールバックされたためマーカーが存在しないことを確認して、安全に再処理します。 このトランザクションバリアントは 受信トレイ パターンであり、生成側の トランザクション送信トレイ パターン の消費側コンパニオンです。

同時重複からの保護

複数の 競合コンシューマーを含む少なくとも 1 回の配信では、2 つのインスタンスが同じメッセージのコピーを同時に受信できます。 どちらもどちらかのコミットの前に存在チェックに合格できるため、チェックだけでは二重処理を妨げることはありません。

アプリケーション ロジックではなく、データ ストアに正確性を適用します。

  • 重複除去キーに対して一意の制約を使用します。 どちらのトランザクションもキーの挿入を試みますが、成功するのは 1 つだけです。 もう 1 つの制約は失敗し、メッセージは重複として扱われます。 このアプローチにより、データベースはレースの単一アービターになります。

  • キャッシュ内でのチェック後設定による競合状態を避ける。 キーをチェックし、2 つの異なる操作に設定するパターンには、両方のキーを要求する同時再試行を可能にするウィンドウがあります。 競合時に失敗する挿入や、存在しない場合の設定操作など、アトミックな条件付き書き込みを使用して、キーの要求が 1 つのアトミック ステップになるようにします。

トランザクションに参加できない副作用を処理する

サードパーティの API の呼び出しや外部ストアへの書き込みなど、コンシューマーのデータベース トランザクションに参加できないプロセスもあります。 これらのプロセスでは、次の 2 段階のアプローチを使用します。

  1. 外部アクションを実行する前に、進行中の状態 で キーを記録します。
  2. プロセスを実行します。
  3. レコードを更新して 完了 し、結果を格納します。

再配信時に、 完了した レコードを使用すると、呼び出しの繰り返しをスキップできます。 進行中のレコードは、以前の試行が部分的に完了したか、別のコンシューマーによって処理されている可能性があることを通知します。

問題と考慮事項

このパターンを実装する方法を決定するときは、次の点を考慮してください。

  • もともと冪等性のある操作を優先してください。 いくつかの操作は本質的に冪等であるため、重複排除のための管理は不要です。 ビジネス識別子をキーとするアップサート、インクリメントではなく絶対値を設定する書き込み、またはリソース識別子に対する HTTP PUT 要求は、1 回実行しても何回実行しても同じ結果になります。

    場合によっては、イベント伝達型状態転送によって、操作を自然にべき等にできます。つまり、メッセージが注文の新しいステータスなどの結果としての絶対状態を含むため、コンシューマーはそれを相対的な変更ではなくアップサートとして適用します。

    Tip

    まず自然に冪等となるように設計し、自然に冪等化できない操作に対してのみ重複排除手法を追加します。

  • 重複除去レコードのライフサイクルを管理します。 重複除去レコードは、期限切れでない限り蓄積されます。 ブローカーが元のメッセージを再配信できる限り、少なくとも各レコードを保持します。 この期間は、ブローカーの最大配信試行回数、ロックまたは可視性のタイムアウト、メッセージの有効期間によって異なります。 この期間を超える重複除去レコードの有効期限を設定して、遅延再配信がそのマーカーを引き続き見つけられるようにします。 レコードを早期に削除すると、ウィンドウが再度開き、重複が発生します。 再送信は通常の再配信期間を大幅に過ぎた後に行われる可能性があるため、デッドレター キューからオペレーターが再送信するメッセージも考慮に入れてください。

  • 手動による重複除去の代わりにメッセージング フレームワークを使用します。 重複除去ストア、アトミック コミット、レコードのクリーンアップを正しく実装すると、エラーが発生しやすくなります。 メッセージ ベースのフレームワークは、このパターンを組み込み機能として提供します。

    たとえば、 NServiceBus はメッセージ識別子によって受信メッセージを重複除去し、重複除去データの構成可能な保持とクリーンアップを提供します。 MassTransit コンシューマー受信トレイは、受信したメッセージをメッセージ識別子で追跡し、コンシューマーがメッセージを一度だけ処理する動作を実現します。

  • ブローカーによる重複排除によって、冪等なコンシューマーロジックの必要性は低減されますが、なくなるわけではありません。 一部のプラットフォームでは、トランスポート層で重複をフィルター処理します。 重複検出Azure Service Bus、構成された時間枠内で繰り返しMessageIdを実行するメッセージを破棄します。これにより、プロデューサー送信の再試行によって発生する重複が抑制されます。 この機能は、送信側および境界付きウィンドウ内で動作します。 再配信後にコンシューマーが同じメッセージを 2 回処理するのを防ぐわけではないので、べき等コンシューマー ロジックが必要です。 プラットフォーム機能は、パターンの代わりではなく、重複するボリュームを減らす防御の第 1 層として扱います。

  • メッセージの順序付けを考慮します。 重複除去では重複が削除されますが、順序は保証されません。 コンシューマーが処理順序に依存している場合は、このパターンを、メッセージ セッションのAzure Service Busなどの順序付けメカニズムと組み合わせるか、コンシューマーが古いメッセージを拒否できるようにするシーケンスまたはバージョン データを含めます。

  • 可観測性のためのインストルメント。 重複除去キーと相関識別子を構造化ログに出力し、検出された重複のメトリックを追跡します。 重複率の上昇は、プロデューサーの構成ミス、受信確認またはロック ウィンドウの不足、または異常なコンシューマーを示している可能性があります。 エンドツーエンドのトレースと相関関係を使用して、サービス間のメッセージに従います。

  • べき等性を下流の呼び出しにも適用します。 1つのコンシューマーを冪等にしても、そのコンシューマーが呼び出すサービスは保護できません。 コンシューマが処理の一環としてダウンストリームサービスを呼び出す場合は、各層で自身の処理を重複排除できるように、べき等キーを伝播します。

このパターンを使用する場合

このパターンは次の状況で使用します。

  • 少なくとも 1 回の配信を提供するブローカーからのメッセージを使用します。これは、ほとんどのブローカーの既定値です。

  • メッセージを再処理すると、財務トランザクションの重複、リソースの作成の重複、通知の繰り返しなど、正しくない結果が生成されます。

  • 複数の競合コンシューマーが同じチャネルを処理するので、重複配信が同時に行われる可能性があります。

このパターンは、次の場合に適さない場合があります。

  • コンシューマーが実行するあらゆる操作は、もともと自然に冪等であるため、再処理しても問題はなく、重複排除のための管理処理は利点がないままコストだけを増やします。

  • ワークロードは、重複する処理の影響を許容でき、重複除去ストアのコストが重複の影響を上回ります。

メッセージングを超えた冪等処理

このパターンはメッセージコンシューマーに冪等性を持たせるものですが、冪等処理はより広い信頼性の原則です。 同じタスクに対して複数回実行できる操作には、その利点があります。 この原則には、再生されたデータを再処理する抽出、変換、読み込み (ETL) 変換、チェックポイントから再開するストリーム処理、重複または再開するスケジュールされたジョブ、重複する配信を受け取る Webhook または HTTP エンドポイントが含まれます。

いずれの場合も、同じコア手法が適用されます。

  1. 安定したキーを使用して作業単位を識別します。
  2. 既に処理した内容を記録します。
  3. 作業を繰り返しても結果が変わらないように、重複をスキップまたは吸収します。

この記事のメカニズム (安定キー、アトミック マーカー、一意の制約など) は、メッセージ ブローカーが関与していない場合でも、それらのコンテキストに転送されます。

ワークロード設計

ワークロードの設計においてべき等コンシューマー パターンをどのように活用して、Azure Well-Architected Framework の柱で扱われている目標と原則に対応するかを評価します。 次の表は、このパターンが各柱の目標をサポートする方法に関するガイダンスを示しています。

柱 このパターンが柱の目標をサポートする方法
信頼性設計の決定は、故障に対するワークロードの回復性を高め、障害の発生後にワークロードを完全な機能状態に回復させるために役立ちます。 このパターンにより、ワークロードは、データを破損させることなく、少なくとも 1 回の配信と安全な再試行を使用できます。これにより、重複配信が正確性リスクから許容条件に変わります。

- RE:07 自己保護
- 一時的な障害の処理

このパターンによって柱内にトレードオフが生じる場合は、他の柱の目標に照らして検討してください。

例

次の例は、Azure Service Bus から受信した注文を処理し、Azure Cosmos DB for NoSQL に状態を永続化する冪等コンシューマーを示しています。

プロデューサーは、Service Bus MessageIdをビジネス レベルの注文識別子に設定します。 コンシューマーは PeekLock モードでメッセージを受信します。このモードでは、コンシューマーがロック期間内にメッセージを完了しなかった場合にメッセージが再配信されます。 注文識別子 (/orderId) 上のコンシューマーのAzure Cosmos DB コンテナー パーティションは、ドキュメント idを同じ注文識別子に設定するため、特定の注文のすべてのコピーが同じ論理パーティションに解決され、注文レコード自体が重複除去マーカーとして機能します。

コンシューマーは、次のように各メッセージを処理します。

  1. メッセージを読み取り、その MessageId を重複除去キーとして使用します。
  2. idとパーティション キーの両方を注文識別子に設定して注文ドキュメントを作成します。
  3. 作成に成功した場合は、Service Bus がそのメッセージをキューから削除するように、メッセージを完了として処理します。
  4. その id を持つドキュメントが既に存在するため、作成が HTTP 409 (競合) 状態で失敗した場合は、既存のドキュメントを読み取り、現在のメッセージと比較します。 格納されている要求ハッシュまたは変更できないビジネス フィールドが一致する場合は、メッセージを重複として扱い、完了し、処理をスキップします。 一致しない場合、プロデューサーが異なるコンテンツに同じ識別子を再利用した可能性や、最初に処理されて以降に注文の詳細が変更された可能性があるため、そのメッセージを黙って破棄するのではなく、デッドレターキューに送るか、アラートを発してください。
  5. 一時的な理由で処理が失敗した場合は、メッセージを破棄してService Bus再配信するか、別のコンシューマーが受け取るようにロックの有効期限が切れます。

作成操作はアトミックであるため、重複除去チェックと書き込みの両方として機能します。 同じメッセージのコピーを受け取る 2 つのコンシューマーは、両方とも注文を作成できません。 一方の作成は成功し、もう一方は競合エラーを返して、重複したものを安全に破棄します。

処理で複数のドキュメントを書き込む必要がある場合は、重複除去ドキュメントと同じパーティション キー内のビジネス ドキュメントの両方を含む トランザクション バッチ を使用します。 トランザクション バッチは 1 つの論理パーティション内で動作するため、1 つのメッセージ共有のすべてのドキュメントのパーティション キーを選択します。 バッチはすべてのドキュメントをまとめてコミットするか、まったくコミットしないため、処理と受信確認の間でクラッシュが発生し、重複除去マーカーとビジネス データが同期しなくなります。既に存在するドキュメントを作成しようとするバッチは、重複を識別する 409 (競合) 状態を返します。

このコンシューマーが重複する送信再試行に対して回復性を持つようにするには、キューで 重複検出 を有効にします。 重複検出は、履歴保持期間内での重複送信を抑止し、冪等コンシューマーは、その期間外に発生する重複や再配信によって生じる重複を処理します。

次のステップ

  • Last updated on