[概要編]」はこちらです。レッドハットでインテグレーションのためのミドルウェア製品のテクニカルサポートを担当している山下です。概要編ではメッセージングの良い面ばかりに焦点を当ててきましたが、今回の疑問編ではメッセージングを検討し始めたときに疑問に思ったり困りがちなことを説明したいと思います。概要編とは異なり、細かな技術的内容も含まれますので、その時々で必要な部分や興味ある部分だけ読んでいただければと思います。(ところで、当初は前回を前編、そして今回を後編にして終わらせようと思っていたのですが、今回もあまりに長くなってしまったので、構成を変えたのでした。このため当初の前編は概要編と名前を変更しています。)ではまず主に疑問とされることを確認して、その後に対処法を見ていきましょう。メッセージは重複配信されることがあるマイクロサービスでは、システムを構成する各種ノードやネットワークで障害が発生することを前提として受け入れます。一方で、それぞれのサービスは個別にデータベースを持ち、機能上の制約、パフォーマンス特性、そして耐障害性などの理由からXAのような分散トランザクションを利用しません。このためメッセージが失われないようにするには、サービスやメッセージングブローカー間の様々なレイヤーにて送信が成功したことが確実に判断できるまでリトライが行われます。一方でこれはメッセージの重複配信が行われる原因ともなります。どのようにサービス内のデータベース更新とメッセージ発行をアトミックに実現するのか?サービスはコマンドを受けると、データベースを更新して同時にイベントを発行することが一般的です。データベースとメッセージングブローカーは、前述のように分散トランザクションは利用しません。このためデータベースのコミットには成功したものの、メッセージ送信には失敗したということがあり得ます。Kafkaはシステムおけるデータハブとしての役割を担っているため、受け入れられないデータ不整合の原因となります。サービスが発行したコマンドに対する応答を受け取る方法RESTなどの同期通信ではその応答結果を受け取ることが可能です。Forgetとも言われるように、メッセージは投げっぱなしとするために、サービスはその応答を受取ることはできません。ビジネス上のアクションや応答を期待してコマンドを送信するケースもあります。多くのモノリシックなシステムではコマンドは同期処理され、応答が返ってくれば処理は完了しており、完了画面に処理結果を表示させることができます。メッセージングでのコマンドは、キュー(トピック)に入り、いつコマンドが処理されるのか、あるいはされたのか、そして処理結果としてビューが最新になっているかどうかも、簡単にはわかりません。ユーザは最新でない情報に基づいた判断によって新規コマンドを発行してしまう可能性もあります。消えないイベントとイベントのバージョニングリレーショナルデータベースであればスキーマはAlter文で変更することができます。イベントを一次情報として永続化するイベントソーシングでは、たとえバグによる間違ったイベントでも原則として消せません。一方ではGDPRにも代表されるように消去はユーザーの権利でさえあり得ます。(後続編にて)モノリシックなシステムとは異なり、マイクロサービスではサービスが協調して全体としてビジネスプロセスが実現されるので、サービス間のプロトコルやそこで利用されるメッセージの設計が重要となります。サービスごとにローカルデータベースが存在するために、サービスをまたいた完全な一貫性を保ったトランザクションは実現できません。ビジネスプロセスは通常複数のサービスにまたがって実現されるため、個別のサービスでのローカルトランザクションを組み合わせて、データの整合をとりながらビジネスプロセスを完了なければなりません。どのように実現したら良いのでしょうか?(これについては別途後続編で説明します)対処方法それでは、これらの疑問に対して対処法を見ていきましょう。そのまま適用できる完全な解決策ばかりではありませんが、可能な限りヒントとなることを記載するようにしました。実際の開発に適用するには参考文献も合わせて参照してください。メッセージは重複配信されることがあるアプリケーションでの冪等性の考慮マイクロサービスではサービス個別にデータベースを持ち、データベースあるいはメッセージングブローカーの機能上の制約や、パフォーマンスそして信頼性の理由から、分散トランザクションは利用されません。これらのサービスやメッセージングシステムの間でメッセージが失われないようにするためには、様々なレベルのレイヤーで送信が成功したことが確実に判断できるまでリトライが行われることになります。once)」の配信となります。once)」の配信となります。リトライの回数や時間に上限を設けるならロストも重複も発生する可能性があります。once)」の配信を前提とし、重複配信時のデータ不整合を防ぐための設計がアプリケーションレベルで必要となります(冪等性の設計)。これはそれぞれのコマンドやイベントごとに考慮しなければならない設計ではあるものの、実際にはそれほど頻繁に重複やロストが起こるわけではありません。多少なりデータ不整合が許容できる場合には対策しなくても良いことがあります。株価情報やIoTによる情報など、絶え間なく最新のデータが送られてきて、ロストしてもすぐ次のメッセージの最新状態で上書きすれば実用上は問題ないといったケースもあります。"いいね"ボタンなど、ロストさせてもビジネス上の大きな影響はなく、ユーザが気がついたときに再実行すればよい程度のケースもあります。重複処理しても結果がうまいこと変わらないといったケースもあります。ユーザ名をユーザ名で重複して上書きしても結果が変わらないといった具合です。これを積極的に用いて、必要とされる最新状態をイベントに含めておき、重複処理されても結果が同じとなるイベントに設計しておく方法もあります。このように無視もしくは回避する方法をまずは検討します。once)」の処理を行わなければならないこともあります。パターンを利用します。基本的な考え方としては、メッセージにトランザクションID(もしくはリクエストID)を持たせるようにします。受信側(コンシューマ側)では同じトランザクションIDを持ったメッセージが配信された場合には重複として破棄するようにします。重複ではないメッセージは、その処理を行うとともに同一ローカルトランザクション内でトランザクションIDを保存します。実装方法にはいくつか方法がありますが、リレーショナルデータベースを利用している場合の単純な方法は次のようなものです。新規に発生したコマンドやイベントにはトランザクションIDとしてUUIDを付与し、そしてそこから後続で派生して発行されるイベントやコマンドにも同じIDを使いまわすようにします。受信側(コンシューマ側)では、オブジェクトID(より正確には集約ルートのID)もしくはコマンド型やイベント型ごとに、処理したメッセージのIDを数日程度保存するようにします。NoSQLやイベントソーシングでは、いくらか異なる方法を用いますが基本的な考え方は同じです。イベント特有の重複排除の方法として、オブジェクトが発行するイベントに(トランザクションIDではなく、)インクリメンタルなシーケンスナンバーをつける方法ががあります。この場合、受信側(コンシューマ側)では、オブジェクトごとに最大のシーケンスナンバーのみを保存すればよく、シーケンスナンバー以下の値を持つメッセージは重複として破棄することができます。これは個別のオブジェクト(より正確には集約)の単位で、適切な排他制御などにより、インクリメンタルなシーケンスナンバーをイベントに付与でき、そして受信側(コンシューマ側)にてイベントの処理順も担保できる場合に有効です。このように重複処理の抑制は、(メッセージングブローカーやサービスメッシュなどの通信の問題として完全に分離することはできないため、)アプリケーションレベルでの冪等性の設計や支援を含めて担保される必要があります。Kafkaの機能[興味ない方は読み飛ばしてください]Kafkaを利用する場合には、条件によっては簡単に重複配信に対応できるケースがあります。単一Kafkaクラスタ内にて、(他のデータベースを更新せずに、)メッセージの送受信とその処理を行えるのであればKafkaのトランザクション機能が使えます。メッセージ内容の変換や拡張をするようなケースで利用できます。Kafkaのメッセージには、トピックのパーティション内での位置を表すオフセット値が含まれており、これをトランザクションIDの代わりに受信側(コンシューマ)で保存して利用する方法もあります。これはトピック内の同一メッセージが受信側(コンシューマ)に重複して配信されるケースには有効です。トピックに対して同一メッセージがリトライなどにより送信されて、実際にメッセージがトピック内に重複して存在してしまった場合には検知できません。オプション)もあります。この仕組みでは送信側(プロデューサ)のKafkaクライアントは、メッセージにシーケンスナンバー(とプロデューサID)をつけるようになります。送信側(プロデューサ)が何らかの理由によりリトライしたことによる重複は、Kafkaブローカー側で検出して破棄するようになります。ところがこの方法も完璧ではなく、送信側(プロデューサ)プロセスに障害が生じ、再起動などによってKafkaへのメッセージ送信そのものが再実行されるような状況では、(新たなプロデューサIDが割り当てられ、)同一メッセージとはみなされずにトピックに重複が発生します。それでも前述のトランザクションIDの仕組みとは併用することができるので、いくらかでも重複を抑制するためにお守りとして有効にしておくことはできます。once)」の配信のために利用できる機能がいくつかあるのですが、条件や制限もあるために、実際に利用できるかどうかは状況次第です。パターン)サービスはコマンドを受けると、データベースを更新して同時にイベントを発行する必要がありますが、これらを同時にXAなどの分散トランザクションに参加させることはできません。パターンが利用できます。これは予めデータベースの中にメッセージを一時的に挿入することができる概念的なキューとしてのOutboxテーブルを用意しておきます。その上で、発行するイベントは、アプリケーションのデータ更新と同じトランザクション内でOutboxテーブルに挿入してコミットします。これによりアプリケーションのデータ更新とイベント発行の両方が確実に行われる(あるいはロールバックにより両方が確実に行われない)ことを保証します。テーブルに挿入されたイベントを取り出してKafkaのトピックへとメッセージを送信します。この際にDebeziumなどのチェンジデータキャプチャ(CDC)を利用することも可能です。outboxテーブルからKafkaへとメッセージを転記する際の障害やプロセスが落ちて再実行されるような場合に、メッセージが重複して発行される可能性があります。このため前述の冪等性の設計も合わせて必要になることに注意してください。RESTのような同期通信でも冪等性の考慮が必要なことがありますところで少し話は戻って、冪等性の考慮はメッセージング特有の問題とは言えなくなってきました。利用により、RESTなどによる同期通信であってもリトライできる(場合によってはロストも防ぐ)環境が整ってきました。このため同期通信のケースでも冪等性の設計が必要になることがあります。同期通信を用いてサービス間をまたぐビジネスプロセスを実現することは少し慎重になったほうが良いかもしれません。SOAでの失敗パターンの1つとして、サービスを横断してビジネスプロセスを実現する中央集権的なプロセス管理がコントロールを手放さず、個別サービスに対して同期通信による主従関係を強いたため、サービスに本来あるべき知識とコントロールを奪って骨抜きにしたことが指摘されています。サービスが発行したコマンドに対する応答を受け取る方法そもそもコマンドに対する応答とは?RESTなどによる同期通信ではその場で応答結果を受け取ることが可能です。メッセージングでは、コマンドに対する明示的な応答というものは基本的に存在しません(しかし例外的に取り扱う方法は後述)。サービスは何らかの状態変更があればイベントを発行します。コマンドに続いて発生する状態変更のイベントを購読することで、遠回しながら結果や状況を知ることができます。RESTなどの同期通信では、コマンドに対する処理が長くとも1分程度で終わることが暗黙的な前提となっており、人を介するプロセスや長時間のビジネスプロセスは発生しませんし、返り値の型や値も通常は1つに定まります。一方の非同期なコマンドではそうした制限はなく、明示的に返り値を受けるための専用の仕組みが存在するわけでもありません。これはサービスが組織と対応するものだと考えれば分かりやすいかもしれません。実際のサービスや組織では、同期通信に求められるような1分間でビジネス上の意味ある結果を提供できる機能ばかりではありません。注文コマンドを考えた場合、コマンドの宛先の組織(サービス)内部では受注生産や発注のために数日かかるビジネスプロセスが実行(更に内部で受注イベントなどの細粒度ともいえるイベントやコマンドが利用)されることもあります。注文コマンド1つに対して、「発送済イベント」や「到着済イベント」といった複数のイベントが時間を分けて発生することもあります。注文内容の不備によるバリデーションエラーや、生産停止や在庫破損による「引当失敗イベント」といった、様々な型を持つ例外イベントが発生する可能性もありますし、当初の呼び出し先とは異なるサービスから発行されたイベントであることもあります。このようにコマンドに対する応答イベントというのは、状況によって時間も型も数も、そして応答元サービスさえも異なるものです。それでも、コマンドとイベントの対応関係は実際のビジネスや組織のインタラクションに対応しており、サービス間において同意を得た明確な契約に従うものです。したサービスの間のコマンドと応答イベントの繰り返しの中でビジネスプロセスが進行していきます。実際にこうしたビジネス上の様々なイベントに対応することは困難ではありますが、むしろそれらへの対応関係を歪めずに直接相対できることが非同期メッセージングの価値でもあります。コマンドに対する応答イベントを受け取る方法イベントはサービスの状態変更をシステム全体に向けて知らせる仕組みなので、コマンドの送信元に向けて専用に発行されるものではありません。逆に言えばコマンドの送信側サービスは、自身の興味ある各種のイベントを監視して備える必要があります。これには様々な方法がありますが、ここでは関係するイベントを受け取ってハンドルするための具体的な方法例をもう少しだけ説明します。通常イベントの中には、変更されたオブジェクトのID(より正確には集約ID)が含まれており、どのオブジェクトに変更があったのかは分かります。イベントを購読するサービスでは、その内部の1つ(または複数)のオブジェクトにイベントをディスパッチすることになります。この際にイベント内容からディスパッチ先のオブジェクトが即座に特定できることもありますが、そうでない場合にはサービス内にある情報を用いてディスパッチ先のオブジェクトを特定しなければなりません。オブジェクトを特定するマッピング方法として、まずは関連する外部キーであるオブジェクトID(あるいはビジネス上の候補キーとなる何らかのNo.など)を利用する方法があります。イベントを購読するサービス内のオブジェクトにて、外部キーとしてオブジェクトのIDを保持している場合には、その関連を利用してマッピングすることができます。他には相関IDを利用する方法もあります。これはメッセージIDや前述のトランザクションID(あるいはそれに類したID)をコマンドに付与し、コマンドから関連して発生したイベントに、それを相関IDとして引き回させ続けることでマッピングします。マッピングを完了させてオブジェクトを特定した後は、即座にイベントハンドラ内でオブジェクトにイベントを処理させることもできますが、受け取ったイベントをコマンドのキュー(トピック)側に入れ直して実行を待つ方法もあります。これは(概要編の最後の方で少し触れたように)排他制御のために複数のスレッドから同じオブジェクトを操作させたくない場合や、1つのイベントから多数のオブジェクトへとマッピングが必要な状況で役に立つことがあります。ところで、これから何らかの作成を指示するようなコマンドでは、そもそも相手先オブジェクトもそのID自体も存在しません。このような場合には、コマンド発行側で相手先オブジェクトのID(UUIDなど)を代理作成して保持しておき、それを作成指示コマンドで指定することで、相手先オブジェクトのIDを事前に把握する方法もあります。いえコマンドの応答を明示的に受けたいコマンドに対する応答をメッセージングで受けられないのはどうしても困るケースがあるかもしれません。同期通信のようにコマンドの実行結果を明示的に受ける方法として、(イベントのトピックを通さずに、)コマンド結果を専門に受けるキュー(トピック)を用意する方法もあります。この場合、サービスのコマンド結果は(イベントとは別に、)この専用のキューに相関IDをつけたメッセージとして応答します。この方法は、特定のコマンドや呼び出し元に特化した結果をメッセージに付加することができ、またコマンドエラーなどで状態変更がない場合でも結果を受け取れます。別途後述するSAGAを実装するためにも役立つこともあります。サービス間の依存関係を強固なものにしてしまう傾向もあるために利用には注意が必要です。した実行結果を単に伝えるようなメッセージは、コマンドやイベントとは役割が異なるため、ドキュメントと呼ばれて区別されることがあります。ユーザがコマンドの実行状況を把握する方法ビューを通して状態を知る多くのモノリシックなシステムではコマンドは同期処理され、応答が返ってきた時には処理は完了し、完了画面に処理結果を表示させる可能です。メッセージングによる非同期なコマンドでは、コマンドはキュー(トピック)に入り、いつコマンドが処理されるのか、あるいはされたのかは簡単には分かりません。コマンドの結果としてユーザに対する明確な応答はなくとも、ビジネスプロセスの進捗に応じて、サービス内のオブジェクト状態が通常は変わっていきます。コマンドの返り値ではなく、ビューからこれらのオブジェクトの状態を参照することで進捗を確認することができます。実際のところメッセージングによるシステムでは、コマンドの実行完了に続いて様々なイベントが発生しながら後続処理進んでいくために、コマンド自体の完了にはあまり意味があるものではありません。(完了画面への遷移というよりは)一覧画面や詳細画面などから様々なオブジェクトのステータスをいつでも確認できるようにします。個別ユーザ向けのメールボックスの提供もしも個別ユーザへの通知が必要なら、各ユーザごとにメールボックスを提供し、そこへコマンド成否や進捗状況を通知する方法もあります。これは実際にメールで通知する方法もあれば、マイページのようなところから汎用的に通知を参照/管理できるメールボックスを用意する方法もあります。個別の要件に特化した各ユーザ向けのビュー(個人の注文履歴や申請履歴など)を開発することもできます。ユーザの利用している画面やデバイスへと通知が必要なら、WebSocketやgRPCのストリームを用いて、コマンドを発行したユーザの画面やデバイスに通知やデータ転送することもできます。データハブとしてのKafkaは状態変化のイベントを様々なサービスに配信することができますが、これは実のところデータセンター内にとどまるものではありません。クライアントの画面やデバイスへと同様にイベントのプッシュ範囲を広げ、デバイス内にビューを作成して画面表示することも可能です。多くのクライアントでは、こうしたビューモデルやイベントを取り扱うための十分なフレームワークやライブラリを既に持っています。クライアント側でのバリデーションところで現実的な問題として、ユーザの実行したコマンドが何らかの不備によってバリデーションによりエラーとなった場合には、やはり即座にユーザへと通知できなければ不便なこともあります。このためクライアント側での事前のバリデーションが特に有効です。これにはコマンド実行の条件を満たさない場合に、クライアント画面側で選択できる数値を制限したりやボタンを押せないようにすることや、WEBサーバで簡易バリデーションを行うことも含まれます。最新データはサービス側にしかないので、こうした事前のバリデーションにも限界があり、サービス側でのバリデーションも合わせて必要です。これによってバリデーションロジックがクライアント側とサービスに分散しますが、いくらかクライアント側へのロジック流出を回避する方法として、ビューにバリデーションを補助する情報を持たせる方法もあります。ビューに、注文の変更が可能かどうか、またキャンセルが可能かどうか、といったフラグを持たせ、クライアント画面ではこのフラグを基にボタンの有効無効を判断するようにします。モノリシックなシステムとは異なり、イベントソーシングの場合には様々な目的に特化したビューをいくらでもカジュアルに作成して利用するようになります。イベントソーシングでは集約を超えた単位でのバリデーション(例えばユーザ名のユニークチェックなど)は難しい傾向もあるため、このような場合にはビューを用いたクライアント側でのバリデーションが重要になります。楽観的ロックで最新情報による判断を強制するメッセージングにおいては、ビューが最新になっているかどうか簡単にはわかりません。ユーザは最新でない情報に基づいた判断によって新規コマンドを発行してしまう可能性や、タイミング悪く他のユーザが既に別コマンドで指示していることもあります。これはメッセージングに特有の問題ではなく、一般的な静的画面のブラウザアプリなどでも起こることですが、こうした場合と同様に楽観的ロックによるバージョンチェックを行って解決することができます。これは予め画面に渡す各レコードや情報にバージョン番号をもたせておきます。コマンド発行時にはそのバージョン番号をあわせて送信し、サービス側では対象レコードのバージョン番号が同じ場合にのみコマンドの処理を行います。バージョン番号が異なっていれば、最新の情報に基づく判断でないか、もしくはタイミング悪く他のユーザが同じオブジェクトを既に更新したとして、コマンドをエラーにします。消えないイベントとイベントのバージョニング間違ったイベントを無効にするリレーショナルデータベースであれば、スキーマはAlter文で変更することもできます。イベントを一次情報として永続化するイベントソーシングでは原則としてイベントは消せません。これはイベントがビジネス上の監査ログの意味合いとなるだけでなく、実装上も一貫してイベントをイミュータブルにすることで、進化を続ける各種ビューを繰り返し導出するために必要なことでもあります。注文を削除するという、ビジネス上考えられることであれば、単に「注文削除イベント」を発行すれば、イベントログにはそれが追記され、そしてそのイベントを購読するビューは、注文に削除フラグを立てるか、あるいは実際にビューのデータベースからレコード削除することができます。バグによって間違って作成されてしまったイベントはどうしたら良いのでしょうか。この場合キャンセルイベントを利用する方法があります。これはキャンセル対象のイベントへのリンクと、キャンセル理由のテキストを含めたキャンセルイベントを作成するというものです。GDPRにも代表されるように個人情報の消去をユーザに求められる場合にも、やはりイベント自体は削除しません。このような場合には予めイベントの暗号化を行っておき、ユーザ情報の復号化に必要な暗号化キーを削除します。復号手段をなくすことで事実上の削除を行います。イベントのバージョンアップ各種のモデルやビューはシステムの成長に応じて進化し続ける一方で、イベントソーシングにおいては、イベントはシステムのライフサイクル全体に渡って基本的には削除も変更も行われません。このためイベントを拡張する場合には、従来のイベントと新しいイベントをシステム内に共存させるためにバージョン管理が必要となります。Schema)を利用する方法があります。この場合フィールドの追加などのいくつかの変更に関しては互換性を得ることができます。フィールド名の変更など、弱いスキーマでは対応できないバージョンアップもあります。この場合には、イベントにバージョン番号をつけて、メッセージの受信側(コンシューマ)にてアプリケーションが理解できるイベントに、ダウンキャストやアップキャストをする仕組みを実装することになります。したイベントの変換コードが重なり続ければ負の遺産になり得ます。基本的にはイベントは削除できないものですが、とはいえ不都合なこともあるかもしれません。System」書籍の中で、上記を含む様々なバージョン更新のテクニックや手順が紹介されていますので興味ある方はご覧ください。まとめ今回の疑問編では、メッセージングに関して良く疑問とされることとその主な解決方法を説明してきました。Consumerパターンについて確認しました。パターンも説明しました。その後、コマンドにおいてそもそも応答とは何か、そしてその応答イベントを受ける方法についても説明しました。ユーザがコマンドの実行状況を知る方法や、関連してクライアント側でのバリデーションが有効であることも確認しました。最後にイベントを無効にするキャンセルイベントのテクニックや、簡単ながらイベントのバージョニングについても触れました。すると非同期メッセージングは面倒くさいと思ったかもしれません。実際面倒くさいのですが、これは非同期メッセージングが同期通信とは異なるレベルで、ビジネス上と技術上の問題に取り組んでいるからでもあります。次回は、結果整合性を保ったトランザクションや、サービス間の連携やビジネスプロセスに焦点を宛てた設計として、SAGAやEventStormingについて説明したいと思います。関連して現在のシステム開発で主流となっているリレーショナルデータベースとデータ中心の設計に対する警告についてもいくらか触れたいと思っています。(とはいえ今回の内容も当初は全く書く気のなかったことばかり、、自分自身今回の記事のスコープがよく分からなくなっていて、次回の内容もでたとこ勝負なのであまり期待しないでください、、次はいつ頃になるかな、、。Thanks勉強会での議論がこの記事の基になっています。Streamsが提供されています。Kafkaの運用の自動化を実現する、オープンソースのStrimziをベースとした機能も含まれています。Hat製品に興味ない方もKubernetesを利用している方はStrimziをぜひ使ってみてください。オブジェクト管理に関して皆さんこんにちは、ソリューションアーキテクトの岡野です。HatでOpenShiftのサポートエンジニアをしている林です。HatでOpenShiftのサポートをしているid:nekopです。各記事は著者の見解によるものでありその所属組織を代表する公式なものではありません。その内容については非公式見解を含みます。スターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。