同じメッセージを複数回処理しても 1 回の処理と同じ効果が得られるように、メッセージ コンシューマーを設計します。 少なくとも 1 回の配信を保証するメッセージング システムは、同じメッセージを複数回配信できます。 重複に対する回復性により、メッセージを再処理しても、重複するレコードが作成されたり、顧客に二重請求されたり、その他の望ましくない影響が生じたりすることはありません。
コンテキストと問題
分散アプリケーションは、通常、直接同期呼び出しを使用するのではなく、メッセージ ブローカーを介して作業を交換します。 Azure Service Bus、Azure Event Hubs、Apache Kafka、RabbitMQ などのほとんどのブローカーは、少なくとも 1 回の配信を提供します。 この保証は、エラーが発生した場合でもメッセージがコンシューマーに到達することを保証しますが、ブローカーが同じメッセージを複数回配信できることを意味します。
分散システム全体で 正確に 1 回 の配信を保証することは現実的ではありません。 厳密に 1 回限りのセマンティクスを使用するブローカーでも、コンシューマーにメッセージを配信したり、ブローカーにデータを書き戻したりするなど、直接制御する操作のみを保証できます。 コンシューマーが外部システムに実装する副作用を制御することはできません。 永続的なソリューションは、重複する配信を排除するのではなく、コンシューマープロセスを正しく処理することです。 重複を無視するコンシューマーと少なくとも 1 回の配信を組み合わせると、 効果的に 1 回の 処理が実現します。
重複は、いくつかのソースから発生する可能性があります。
プロデューサーが再試行します。 プロデューサーはメッセージを送信し、一時的なネットワーク 障害またはタイムアウトのために受信確認を受信せず、メッセージを再度送信します。 送信が初めて成功した場合でも、ブローカーは 2 つのコピーを保持するようになりました。
受信確認が見つからない後に再配信します。 コンシューマーはメッセージを受信して処理しますが、コンシューマーがクラッシュしたり、ロックが期限切れになったり、受信確認が失われたりしたため、メッセージの受信確認に失敗します。 ブローカーは、メッセージが処理されなかったと想定し、再度配信します。
処理中のコンシューマーエラー。 コンシューマーはデータベースの書き込みを完了しますが、メッセージを確認する前にクラッシュします。 別のインスタンスがメッセージを選択し、書き込みを繰り返します。
ソリューション
コンシューマーが正常に処理したメッセージの記録を保持し、再配信されても変わらない安定した識別子に基づいて、すでに処理したメッセージをスキップすることで、べき等コンシューマーを実現します。 コンシューマーは、永続的な識別子ストアを調べて、その識別子が既に処理されているかどうかを判断してから、メッセージを処理するか、重複として破棄します。
コア フローは、次の手順で構成されます。
- メッセージを読み取り、その重複除去キーを抽出します。
- そのキーに対応する重複排除ストアを確認します。
- キーがストアに既に存在する場合は、メッセージを重複として扱います。 メッセージを確認して処理を停止し、必要に応じて、以前に記録した結果を返します。
- キーがストアにまだ存在しない場合は、メッセージを処理し、1 つのアトミック操作でキーを記録してから、メッセージを確認します。
次のセクションでは、コンシューマーを冪等にするための指針を示します。
安定した重複除去キーを選択する
重複除去キーは、再配信のたびに論理メッセージを一意かつ一貫して識別する必要があります。 プロデューサーによって割り当てられたメッセージ識別子、または特定の論理操作を識別するビジネスレベルの冪等性キーのいずれかを使用し、複数のメッセージで共有される相関コンテキストは使用しないでください。
たとえば、Service Bus MessageId プロパティを、論理メッセージを一意に識別する値に設定します。
CorrelationIdをキーとして使用しないでください。これは、要求とその応答など、関連するメッセージのグループを参照するためです。 CloudEvents 仕様に従うイベントの場合、 source 属性と id 属性の組み合わせによってイベントが一意に識別され、再配信間で安定した状態が維持されます。
再配信時にブローカーが再生成するトランスポート レベルの識別子や、配信の試行から派生した値にはキーを設定しないでください。 これらの値は配信の間で変化し、重複検出を無効化します。 また、受信タイムスタンプなどの揮発性フィールドからキーを派生しないようにします。
パブリッシュ/サブスクライブ設計の複数のサブスクライバーなど、複数の独立したコンシューマーが同じチャネルを処理する場合、各コンシューマーはメッセージの独自のコピーを受け取り、処理の完了を個別に追跡する必要があります。 コンシューマーが重複除去ストアを共有する場合は、コンシューマー ID とメッセージ ID の複合レコードにキーを設定します。 メッセージ識別子のみをキーとするストアでは、最初のコンシューマーが他のすべてのコンシューマーによる処理を抑止できてしまいます。
処理されたキーを格納する場所を決定する
処理されたキーには、次のストレージ オプションが一般的です。
専用の重複除去テーブル。 コンシューマーは、処理されたキーごとに 1 行を保持する別のテーブル ( 受信トレイと呼ばれることもあります) を保持します。 この方法では、重複除去の問題がビジネス データから分離され、多数のメッセージの種類が同じメカニズムを共有している場合に適しています。
ビジネス エンティティ自体。 コンシューマーは、メッセージが作成または更新するレコードにキーを格納します。 この方法では、別のテーブルを回避しますが、重複除去をビジネス データ型に結合します。
処理されたキーと副作用をアトミックにコミットする
チェックしてから処理するフローには、失敗が発生しうる時間的な隙間があります。 コンシューマーがメッセージを処理し、別の手順でキーを記録すると、2 つの操作間のクラッシュによって副作用は適用されますが、キーは記録されないため、コンシューマーは次の再配信を再処理します。
重複除去マーカーとビジネス副作用を同じトランザクションに書き込むことで、このエラー ウィンドウを回避します。 コンシューマーが両方の操作を一緒にコミットするか、まったくコミットしなかった場合、再配信時にマーカーを見つけてメッセージをスキップするか、トランザクションが完了せず、メッセージを安全に再処理していないため、マーカーが見つかりません。 このトランザクション型のバリアントは Inbox パターンであり、プロデューサー側の Transactional Outbox pattern に対になるコンシューマー側のパターンです。
同時重複からの保護
同時 競合コンシューマーによる少なくとも 1 回の配信では、2 つのインスタンスが同じメッセージのコピーを同時に受信できます。 どちらのインスタンスも、どちらのインスタンスもコミットする前に存在チェックに合格できるため、チェックだけでは二重処理が妨げることはありません。
次の手順を実行して、アプリケーション ロジックではなくデータ ストアに正確性を適用します。
重複除去キーに対して一意性制約を使用して、2 つのトランザクションがキーの挿入を試みることができるが、成功できるのは 1 つだけになるようにします。 もう 1 つのトランザクションは制約に失敗し、メッセージを重複として扱います。 この方法により、データベースが競合の単一のアービターになります。
キャッシュ内でのチェック後設定による競合状態を避ける。 キーをチェックし、2 つの異なる操作で設定するパターンには、同時再試行でキーを要求できるウィンドウがあります。 キーの取得を単一のアトミックなステップにするには、競合時に失敗する挿入処理や存在しない場合に設定する操作などのアトミックな条件付き書き込みを使用します。
トランザクションに参加できない副作用を処理する
サードパーティの API の呼び出しや外部ストアへの書き込みなど、コンシューマーのデータベース トランザクションに参加できないプロセスもあります。 これらのプロセスには、次の 2 フェーズ アプローチを使用します。
- 進行中の状態でキーを記録し、外部アクションを実行します。
- レコードを更新して 完了 し、結果を格納します。
再配信時に、 完了した レコードは、呼び出しの繰り返しをスキップするようにコンシューマーに指示します。 進行中のレコードは、以前の試行が部分的に完了したか、別のコンシューマーによって処理されている可能性があることを通知します。 コンシューマーは、再配信を確認する前に、古いレコードを調整するか、未解決のケースをルーティングして介入する必要があります。
問題と考慮事項
このパターンを実装する方法を決定するときは、次の点を考慮してください。
もともと冪等性のある操作を優先してください。 ある操作は本質的にべき等であるため、重複排除のための管理は不要です。 ビジネス識別子にキーを設定したアップサート、増分ではなく絶対値を設定する書き込み、またはリソース識別子に対する HTTP
PUTは、1 回または複数回実行しても同じ結果を生成します。場合によっては、イベントに状態を持たせる方式を使って、操作を自然に冪等にできます。 メッセージには、注文の新しいステータスなど、結果として得られた完全な状態が含まれているため、コンシューマーはそれを相対的な変更としてではなく、アップサートとして適用します。
Tip
可能であれば、操作自体が自然に冪等になるように設計し、自然な形で冪等にできない操作に対してのみ重複排除の手法を使用します。
重複除去を構成する代わりに、メッセージング フレームワークを使用します。 重複除去ストレージ、コミット、クリーンアップを正しく実装すると、エラーが発生しやすくなります。 メッセージ ベースのフレームワークは、このパターンを組み込み機能として提供します。
たとえば、 NServiceBus はメッセージ識別子によって受信メッセージを重複除去し、重複除去データの構成可能な保持とクリーンアップを提供します。 MassTransit コンシューマー アウトボックスは、受信したメッセージをメッセージ識別子に基づいて追跡し、コンシューマーの exactly-once 動作を実現します。
重複除去レコードのライフサイクルを管理します。 重複除去レコードは、有効期限を設定しない限り蓄積されます。 ブローカーが元のメッセージを再配信できる限り、少なくとも各レコードを保持します。 このウィンドウのサイズは、ブローカーの最大配信試行回数、ロックまたは可視性のタイムアウト、およびメッセージの有効期間によって異なります。
この期間を超える重複除去レコードに対して Time to Live を設定して、遅延再配信がそのマーカーを引き続き検出するようにします。 レコードを早期に削除すると、ウィンドウが再度開き、重複が発生します。 オペレーターがデッドレターキューから再投入するメッセージは、通常の再配信ウィンドウが終了してからかなり時間がたって行われる可能性があるため、考慮に入れてください。
冪等コンシューマーロジックの代わりに、ブローカー側の重複排除を使わないでください。 一部のプラットフォームでは、トランスポート層で重複をフィルター処理します。 たとえば、重複検出Service Busは、構成された時間枠内に
MessageIdを繰り返すメッセージを破棄し、プロデューサー送信の重複再試行を抑制します。この機能は、送信側および境界付きウィンドウ内で動作するため、再配信後にコンシューマーが同じメッセージを 2 回処理することを防ぐことはありません。 引き続き冪等なコンシューマーロジックが必要です。 べき等コンシューマー パターンの代わりにではなく、重複するボリュームを減らすためにプラットフォーム機能を使用します。
メッセージの順序付けを考慮します。 重複除去では重複が削除されますが、順序は保証されません。 コンシューマーが処理順序に依存している場合は、このパターンを Service Busメッセージ セッションなどの順序付けメカニズムと組み合わせるか、シーケンスまたはバージョン データを含め、コンシューマーが古いメッセージを拒否できるようにします。
可観測性のためのインストルメント。 重複除去キーと相関識別子を構造化ログに出力し、検出された重複のメトリックを追跡します。 重複率の上昇は、プロデューサーの構成ミス、受信確認またはロック ウィンドウの不足、または異常なコンシューマーを示している可能性があります。 分散トレースと相関関係を使用して、サービス間でメッセージをフォローします。
べき等性を下流の呼び出しにも適用します。 メッセージコンシューマーを冪等にしても、それが呼び出すサービスは保護されません。 コンシューマーが処理の一環としてダウンストリーム サービスを呼び出す場合は、各サービス層がそれぞれの処理を重複排除できるよう、べき等キーを伝播します。
このパターンを使用する場合
このパターンは次の状況で使用します。
少なくとも 1 回の配信を提供するブローカーからのメッセージを使用します。これは、ほとんどのブローカーの既定値です。
メッセージを再処理すると、財務トランザクションの重複、リソースの作成の重複、通知の繰り返しなど、正しくない結果が生成される可能性があります。
複数の競合コンシューマーが同じチャネルを処理するので、重複配信が同時に行われる可能性が高くなります。
このパターンは、次の場合に適さない場合があります。
コンシューマが実行する操作はもともと冪等であるため、再処理しても問題はなく、重複排除のための管理処理は利点がないのにコストだけを増やします。
ワークロードは、重複する処理の影響を許容でき、重複除去ストアのコストが重複の影響を上回ります。
メッセージングを超えた冪等処理
このパターンはメッセージコンシューマーに冪等性を適用するものですが、冪等処理は、同一のタスクに対して複数回実行されるあらゆる処理に恩恵をもたらす、より広範な信頼性の原則です。 この原則には、再生されたデータを再処理する抽出、変換、読み込み (ETL) 変換、チェックポイントから再開するストリーム処理、重複または再開するスケジュールされたジョブ、重複する配信を受け取る Webhook または HTTP エンドポイントが含まれます。
いずれの場合も、同じコア手法が適用されます。
- 安定したキーを使用して作業単位を識別します。
- 処理内容を記録します。
- 重複する実行をスキップまたは吸収して、作業を繰り返しても結果が変わらないようにします。
この記事のメカニズム (安定キー、アトミック マーカー、一意性制約など) は、メッセージ ブローカーが関与していない場合でも、これらのコンテキストに転送されます。
ワークロード設計
Azure Well-Architected Framework の各柱で取り上げられている目標と原則に対応するため、ワークロードの設計においてべき等コンシューマー パターンをどのように活用すべきかを検討してください。 次の表は、このパターンが各柱の目標をサポートする方法に関するガイダンスを示しています。
| 柱 | このパターンが柱の目標をサポートする方法 |
|---|---|
| 信頼性設計の決定は、故障に対するワークロードの回復性を高め、障害の発生後にワークロードを完全な機能状態に回復させるために役立ちます。 | このパターンにより、ワークロードは、データを破損させることなく、少なくとも 1 回の配信と安全な再試行を使用できます。これにより、重複配信が正確性リスクから許容条件に変換されます。 - RE:07 自己保護 - 一時的な障害 |
このパターンによって柱内にトレードオフが生じる場合は、他の柱の目標に照らして検討してください。
例
次の例では、Service Bus からの注文を処理し、Azure Cosmos DB for NoSQL に状態を永続化する冪等なコンシューマーについて説明します。
- プロデューサーは、Service Bus
MessageIdをビジネス レベルの注文識別子に設定します。 - コンシューマーは PeekLock モードでメッセージを受信します。これにより、コンシューマーがロックの有効期限が切れる前にメッセージが解決されない場合に、メッセージを再配信できるようになります。
- コンシューマーのAzure Cosmos DBコンテナー パーティションは、
/orderId注文識別子で、ドキュメントidを同じ順序識別子に設定するため、特定の注文のすべてのコピーが同じ論理パーティションに解決され、id自体の順序が重複除去マーカーとして機能します。
コンシューマーは、各メッセージを処理するために次の手順を実行します。
- メッセージを読み取り、その
MessageIdを重複除去キーとして使用します。 -
idキーとパーティション キーの両方が注文識別子に設定された注文ドキュメントの作成を試みます。 - 作成が成功した場合は、メッセージを完了してService Busキューから削除します。
- その
idを持つドキュメントが既に存在するため、HTTP 409 (競合) 状態で作成が失敗した場合は、既存のドキュメントを読み取り、現在のメッセージと比較します。 - 格納されている要求ハッシュまたは変更できないビジネス フィールドが一致する場合は、メッセージを重複として扱い、完了し、さらに処理をスキップします。
- 格納されているリクエストハッシュまたは不変のビジネスフィールドが一致しない場合は、メッセージを黙って破棄するのではなく、配信不能キューに送信し、アラートを発報します。 プロデューサーが異なるコンテンツの識別子を再利用した場合や、注文が最初に処理されてからメッセージの詳細が変更された可能性があります。
- 一時的な理由で処理が失敗した場合は、メッセージを破棄してService Bus再配信するか、別のコンシューマーが受け取ることができるようにロックの有効期限が切れます。
作成操作はアトミックであるため、重複除去チェックと書き込み操作の両方として機能します。 同じメッセージのコピーを受け取る 2 つのコンシューマーは、両方とも注文を作成できません。 1 つの作成試行は成功し、もう 1 つの試行では競合エラーが返され、重複したものは安全に破棄されます。
処理で複数のドキュメントを書き込む必要がある場合は、重複除去キーと同じパーティション キー内のビジネス ドキュメントの両方を含む トランザクション バッチ を使用します。 トランザクション バッチは 1 つの論理パーティション内で動作するため、1 つのメッセージ共有のすべてのドキュメントを含むパーティション キーを選択します。 バッチでは、すべてのドキュメントがまとめてコミットされるか、まったくコミットされないかのいずれかであるため、処理と確認応答の間でクラッシュが発生しても、重複除去マーカーと業務データの整合性が失われることはありません。既に存在するドキュメントを作成しようとするバッチは 409（Conflict）ステータスを返し、これにより重複であることを特定できます。
このべき等コンシューマーに、重複する送信再試行に対する回復性も持たせるには、キューで重複検出を有効にします。 Standard または Premium キューでは、重複検出によって履歴ウィンドウ内で繰り返し送信が抑制されます。 べき等コンシューマーは、そのウィンドウの外側にある、または再配信の結果として生じる重複を引き続き処理します。
次のステップ
- 同一入力に対する Azure Functions の設計 では、重複する呼び出しに耐えられるべき等関数を構築するためのガイダンスを提供します。
関連するリソース
Azureの非同期メッセージング オプションでは、配信の保証と重複処理の要件を決定するメッセージング インフラストラクチャの選択肢について説明します。
トランザクショナル アウトボックス パターンは、ビジネスデータと同じトランザクションでメッセージをコミットすることでそれらを確実に発行するパターンであり、べき等コンシューマー パターンのパブリッシャー側に当たります。
再試行パターンを使用すると、アプリケーションは操作を再試行することによって一時的な障害を処理できます。これにより、再試行によって配信が重複する可能性があるため、べき等処理が必要になります。
回復性Azure Event HubsとAzure Functions設計では、イベント ストリームの重複除去手法など、Event Hubs によってトリガーされる関数にこのパターンが適用されます。