RocketMQ message integration

はじめに

Apache RocketMQ は誕生以来、10 年以上にわたる大規模ビジネスでの安定性の実証を経て、Alibaba グループの社内ビジネスの 100% と、Alibaba Cloud の数多くの企業顧客に利用されてきました。金融レベルの信頼性を備えたビジネスメッセージソリューションとして、RocketMQ は当初からビジネス統合分野における非同期通信機能の構築に注力しています。

本記事では、ビジネスメッセージ統合のシナリオに沿って、RocketMQ のタイミングメッセージ機能について、ユースケース、適用事例、機能原理、ベストプラクティスの観点から紹介します。

以下のリンクからライブ解説をご覧いただけます。

https://yqh.aliyun.com/live/detail/29063

概念：スケジュールされたメッセージとは

ビジネスメッセージ統合のシナリオにおいて、タイミングメッセージとは、プロデューサーがメッセージキューにメッセージを送信した後、コンシューマーに即座に消費されるのではなく、指定された時刻にコンシューマーが消費できるようにする仕組みです。

同様に、遅延メッセージもタイミングメッセージの別の解釈であり、プロデューサーがメッセージを一定時間遅らせてからコンシューマーに消費させることを意味します。現在時刻に一定の遅延時間を加えたタイミング指定と理解できます。

通常メッセージとタイミングメッセージのフローを比較します。通常メッセージは、メッセージ送信、メッセージストレージ、メッセージ消費の 3 つのプロセスに大別できます。Topic に送信されたメッセージは、直ちにコンシューマーによる消費待ちの状態になります。

タイミングメッセージや遅延メッセージは、通常メッセージに定期配信機能を追加したものと捉えられます。プロデューサーがタイミングメッセージを送信すると、メッセージは直ちにはユーザーの Topic に入らず、RocketMQ によって一時的にシステム Topic に保存されます。設定された時刻になると、RocketMQ はメッセージを本当の Topic に転送し、コンシューマーが消費できるようになります。

シナリオ：タイミングメッセージが必要な理由

分散タイミングスケジューリングトリガーやタスクタイムアウト処理などのシナリオでは、正確かつ信頼性の高いタイミングイベントトリガーを実現する必要があります。このようなタイミングイベントのトリガーには、一般的に以下の要件があります。

・高いスループット：大量のイベントをトリガーでき、パフォーマンスボトルネックが生じないこと。

・高い信頼性と再処理性：イベントトリガーの消失がないこと。

・分散スケーラビリティ：単一マシンに依存せず、複数のサービスノードにスケジューリング負荷を分散できること。

従来のタイミングスケジューリングは、データベースのタスクテーブルスキャン機構に基づいて実装されることが多く、固定時刻にトリガーすべきタスクをデータベースに格納し、マイクロサービスアプリケーションが定期的にデータベースをスキャンしてタスクを取得・処理する方式です。

この方式でもタイミングスケジューリングは実現できますが、以下のような課題があります。

・重複スキャン：分散マイクロサービスアーキテクチャでは、各マイクロサービスノードがデータベースをスキャンするため、冗長なタスク処理が発生し、再処理が必要になります。

・不正確な固定時間間隔：定期スキャンの仕組みでは、任意の時間精度での遅延スケジューリングを実現できません。

・低い水平スケーラビリティ：重複スキャンの問題を回避するため、データベーステーブルスキャン方式ではサービスノードごとにテーブルを分割しますが、各テーブルは単一ノードでしか処理できず、パフォーマンスボトルネックが発生します。

このようなタイミングスケジューリングのシナリオで RocketMQ のタイミングメッセージを利用することで、スケジューリングタスクの開発ロジックを簡素化し、高パフォーマンスでスケーラブル、かつ信頼性の高いタイミングトリガー機能を実現できます。

・高い精度と低い開発コスト：メッセージ通知ベースの固定ステップ間隔に依存せず、任意の精度でイベントを簡単にトリガーでき、ビジネス側の重複排除処理も不要です。

・高パフォーマンスなスケーラビリティ：従来のデータベーススキャン方式は複雑で、頻繁なインターフェイススキャン呼び出しが必要でパフォーマンスボトルネックが生じやすいですが、メッセージキュー RocketMQ のタイミングメッセージは高い同時実行性と水平スケーラビリティを備えています。

ケース：タイミングメッセージによる金融決済タイムアウト処理

タイミングメッセージを利用することで、一定時間後に特定の操作を実行でき、ビジネスシステム側でタイミング状態を管理する必要がありません。代表的なケースとして金融決済のタイムアウトがあります。注文システムで、ユーザーが注文してから 30 分後に注文状況を確認し、支払いが行われていない場合は自動的に注文をキャンセルするシナリオを考えます。

RocketMQ のタイミングメッセージに基づき、ユーザーの注文時に 30 分後を指定したタイミングメッセージを送信し、MessageKey に注文 ID を設定します。30 分後、注文システムはメッセージを受信して注文 ID で注文状況を確認します。タイムアウト後に支払いが行われていなければ、注文は自動的にクローズされます。

原理：RocketMQ のタイミングメッセージの実装方法

固定間隔のタイミングメッセージ

前述のとおり、タイミングメッセージの核心は、システムタイミング Topic 内のメッセージを特定の時刻にユーザーの Topic にどのように転送するかです。

Apache RocketMQ 4.x のタイミングメッセージは、タイミングメッセージを DelayLevel ごとにシステム内の異なるキューに配置し、各キューに対して定期的にメッセージをプルしてユーザーの Topic に転送する定期タスクを起動します。この方式は実装がシンプルな一方、特定の DelayLevel のタイミングメッセージのみサポートするという制約があります。

現在、任意の秒数に対応するタイミングメッセージの実装がコミュニティに提案されています。ここではその基本的な実装原理を簡単に紹介します。

タイマーホイールアルゴリズム

具体的な実装原理に入る前に、タイミングメッセージ実装のコアアルゴリズムであるタイマーホイールアルゴリズムを紹介します。

上図に示すように、これは 7 秒周期のタイマーホイールで、タイミングの最小精度は秒です。タイマーホイール上には現在時刻を指すポインターがあり、定期的に次の目盛りに移動します。

1 秒後にスケジュールする場合、データを「1」の目盛りに配置します。同じ時刻に複数のデータがある場合は、連結リストの形式で追加されます。

タイマーホイールが「1」の目盛りに回転すると、リストから読み出して処理されます。タイマーホイールの 1 周を超える時刻を指定する場合、たとえば 14 秒後を指定する場合、1 周は 7 秒なので「6」の目盛りに配置します。1 周目が「6」に回転した時点では現在時刻が目標時刻に達していないため、このデータはスキップされます。2 周目が「6」に回転したときに、目標の 14 秒に達したと判断されます。

任意秒数のタイミングメッセージ

RocketMQ では、TimerWheel を使用して時間輪を表現・保存し、アペンドオンリーの TimerLog を使用して時間輪上の各目盛りに対応するすべてのメッセージを記録します。

TimerLog はタイミングメッセージの重要なメタデータを記録し、タイミング時刻後にメッセージをユーザーの Topic に転送するために使用されます。主な属性は以下のとおりです。

TimerWheel は固定長の配列として抽象化でき、配列の各セルが時間輪上の「目盛り」を表します。TimerWheel の「目盛り」には以下のプロパティがあります。

TimerWheel と TimerLog の直接的な関係を以下の図に示します。

TimerWheel の各セルが時間の目盛りを表します。firstPos はその目盛りの最初の TimerLog レコードのアドレスを指し、lastPos は最後の TimerLog レコードのアドレスを指します。同じ目盛り内のメッセージの TimerLog は、prevPos を通じて連結リストとして繋がれています。

新しいレコードを追加する場合、たとえば「1-4」を追加する場合、新しいレコードの prevPos を現在の lastPos（つまり「1-3」）に向け、lastPos を「1-4」に変更します。これにより、同じ目盛り上の TimerLog レコードがすべて連結されます。

TimerWheel と TimerLog を使って、スケジュールされたメッセージが RocketMQ に送信されてから最終的にユーザーに配信されるまでの流れを見ていきましょう。

まず、ユーザーがタイミングメッセージを送信すると、RocketMQ は実際にはそのメッセージをタイミングメッセージ処理専用のシステム Topic に送信します。

その後、TimerMessageStore 内の 5 つのサービスが分担・連携して処理を行いますが、全体の流れは「時間輪への投入」と「時間輪からの取り出し」の 2 つのフェーズに分けられます。

時間輪への投入：

・TimerEnqueueGetService はシステムタイミング Topic からメッセージをプルし、enqueuePutQueue に格納して TimerEnqueuePutService の処理を待ちます。

・TimerEnqueuePutService は TimerLog レコードを構築し、時間輪の対応する目盛りに配置します。

時間輪からの取り出し：

・TimerDequeueGetService は時間輪を回転させ、現在時刻の目盛りのすべての TimerLog レコードを取り出して dequeueGetQueue に格納します。

・TimerDequeueGetMessageService は TimerLog レコードに基づいて CommitLog からメッセージを読み取ります。

・TimerDequeuePutMessageService はキュー内のメッセージが期限切れかどうかを判定します。期限切れの場合はユーザーの Topic に転送して消費を待ち、まだ期限切れでない場合はシステムタイミング Topic に戻して時間輪への再投入を待ちます。

実戦：タイミングメッセージの使い方

RocketMQ の秒レベルタイミングメッセージの原理を理解したところで、実際にタイミングメッセージの使い方を見ていきましょう。まず「タイミング/遅延メッセージ」型の Topic を作成する必要があります。これはコンソールまたは CLI コマンドで作成できます。

上記のとおり、タイミングメッセージはメッセージ送信時に「仕掛けを施す」ものです。プロデューサー側では、送信時に配信予定時刻を設定するだけで、それ以外は通常のメッセージ送信と同じです。

スケジュール時刻に達すると、このメッセージは通常のメッセージとしてユーザーの Topic に配信されます。コンシューマー側では、通常のメッセージ消費と何ら変わりありません。

注意：タイミングメッセージの実装ロジックは、タイミングストレージを経由してトリガーを待ってからコンシューマーに配信されます。そのため、大量のタイミングメッセージのタイミング時刻を同じ時刻に設定すると、その時刻に同時に大量のメッセージを処理する必要が生じ、システムに大きな負荷がかかります。したがって、大量のメッセージに同じトリガー時刻を設定しないことが推奨されます。