Details and Best Practices of RocketMQ Client Load Balancing Mechanism

はじめに

本記事では、RocketMQ の負荷分散メカニズムについて解説します。主に負荷分散のトリガータイミング、クライアント側の負荷分散が消費に与える影響（メッセージ蓄積や消費バーストなど）を取り上げ、ベストプラクティスに関する推奨事項も紹介します。

負荷分散の意義

上図は RocketMQ のメッセージストレージモデルを示しています。メッセージはキューパーティションごとに順序立てて格納されます。RocketMQ のキューモデルでは、プロデューサー、コンシューマー、読み取り/書き込みキューが多対多のマッピング関係を形成でき、無限に拡張可能です。RabbitMQ などの従来のメッセージキューと比べて大きな利点があります。特にストリーミング処理シナリオでは、同一キュー内のメッセージが同一コンシューマーによって処理されることが保証されるため、バッチ処理や集約処理との親和性が高く、有利に機能します。

あるトピックのメッセージを消費するコンシューマーは、そのトピックのすべてのキューのメッセージを消費することになります（上図では、コンシューマー A1 がキュー 1 を消費し、コンシューマー A2 がキュー 2 と 3 を消費しています）。

したがって、各コンシューマーの負荷をできる限り均等にする必要があります。つまり、これらのコンシューマーに同じ数のキューを割り当てると同時に、異常時（クライアントのダウンタイムなど）にはキューを異なるコンシューマー間で移行できるようにする必要があります。

負荷分散メカニズムの分析

負荷分散のタイミング

負荷分散はクライアントとサーバーが連携して行うプロセスです。まず、サーバーとクライアントそれぞれの役割を整理し、最初の問いに答えていきます。負荷分散はいつ発生するのか。

・クライアント側のアクティブな負荷分散

上図は RocketMQ クライアントの関連クラスの構造を示しています。MQClientInstance はサーバーとの通信や、負荷分散を含む基盤サービスの連携を担っています。

MQClientInstance には rebalanceImmediately と doRebalance という 2 つの関連メソッドがあります。負荷分散のタイミングを分析するために、これら 2 つのメソッドがいつ呼び出されるかを確認します。

1. 起動時に直ちに負荷分散を実行する。

2. 一定間隔（デフォルトで 20 秒）で負荷分散を実行する。

・サーバーからの通知による負荷分散

サーバーは MQClientInstance#rebalanceImmediately メソッドを通じて、クライアントに負荷分散の実行を通知します。サーバー側のコードでも関連する呼び出しを確認できます。

上記のメソッドを分析した結果、以下のシナリオでサーバーが積極的にクライアントに負荷分散のトリガーを通知することがわかります。

1. クライアントのオンラインとオフライン

・オンライン

a. 新しいクライアントがサーバーにハートビートを送信する。

・オフライン

b. クライアントがサーバーにオフラインリクエストを送信する。

c. 基盤レイヤーの接続異常：netty チャンネルの IDLE/CLOSE/EXCEPTION イベントに応答する。

2. サブスクリプション関係の変更：新しいトピックのサブスクライブ、または古いトピックのサブスクライブ解除。

負荷分散ポリシー

前述の通り、負荷分散は実際にはコンシューマーが処理を担当するキューの数を変更するプロセスです。ここで、毎回変更が必要なキュー数と影響を受けるクライアント数は、負荷分散ポリシーによって決まります。

一般的な負荷分散ポリシーを分析していきましょう。

・均等分散

AllocateMessageQueueAveragely はデフォルトの負荷分散ポリシーです。

4 つのクライアントと 24 のキューがあり、2 番目のクライアントがオフラインになった場合：

デフォルトの負荷分散ポリシー（AllocateMessageQueueAveragely）を例にとると、再割り当てされるキュー数は 8 です。

デフォルトの負荷分散ポリシーは各クライアントにできるだけ均等にキューを分散できますが、ロードバランシングのたびに再割り当てされるキュー数が多くなります。特にクライアント数が多いシナリオではその傾向が顕著です。

・コンシステントハッシュ

コンシステントハッシュアルゴリズムに基づく負荷分散ポリシー（AllocateMessageQueueConsistentHash）は、負荷分散のたびに再割り当てするキュー数を最小限に抑えますが、負荷が不均等になる可能性があります。

負荷分散が消費に与える影響

実際のオンライン環境の例を見ていきましょう。

下図で、緑の線は送信 TPS を、黄色の線は消費 TPS を示しています。21:00 と 21:50 の前後に消費バーストが発生していることが確認できます。

これら 2 つのタイムポイントはアプリケーションのリリース中です。前述の分析の通り、あるコンシューマーがオフラインになった後、同一グループ内の他のコンシューマーがこの変更を感知するまでにある程度の時間がかかるため、秒レベルの消費遅延が発生します。リリース後、コンシューマーは蓄積されたメッセージを迅速に処理し、消費速度が大幅に上昇していることが確認できます。

この例から、オフライン時には負荷分散による短いメッセージ処理遅延が発生し、新しいコンシューマーはサーバーから消費進捗を取得してから消費を再開することがわかります。コンシューマーが異常停止したり、shutdown を呼び出さずにグレースフルシャットダウンを行わなかった場合、最新の消費ポイントがアップロードされないため、新たに割り当てられたコンシューマーによる重複消費が発生します。

ここで、負荷分散が消費に与える影響をまとめます。クライアントで負荷分散がトリガーされた場合：

1. 新たに割り当てられたキューで重複消費が発生する可能性がある。これは公式ドキュメントで消費の冪等性が求められている理由でもある。

2. 消費担当から外れたキューは一時的に停止する。元の消費 TPS が非常に高い場合や生産ピークと重なった場合、消費バーストを引き起こす。

ベストプラクティス

クライアントの頻繁なオンライン/オフラインを避ける

負荷分散の影響を避けるため、クライアントのオンライン/オフラインをできる限り減らし、消費の冪等性を確保する必要があります。

同時に、アプリケーションの再起動やオフライン前には shutdown メソッドを呼び出す必要があります。これにより、サーバーがクライアントからのオフラインリクエストを受信後、速やかに他のクライアントに負荷分散のトリガーを通知し、消費遅延を削減できます。

適切な負荷分散ポリシーの選択

ビジネス要件に応じて負荷分散ポリシーを柔軟に選択する必要があります。

・クライアント間の負荷をできる限り均等にする必要がある場合：デフォルトの均等分散ポリシーを選択する。

・アプリケーション再起動による消費遅延を削減する必要がある場合：コンシステントハッシュの割り当てポリシーを選択する。

他の負荷分散ポリシーについては時間の関係で個別の紹介は省略するため、読者自身での探求をお勧めします。

クライアントのサブスクリプションの一貫性を確保する

RocketMQ の負荷分散は各クライアントが独立して計算するため、各クライアントの負荷分散アルゴリズムとサブスクリプション文の一貫性を確保する必要があります。

・負荷分散ポリシーの不一致は、複数のクライアントに同一キューが割り当てられたり、一部のクライアントがキューを割り当てられない原因となります。

・サブスクリプション文の不一致は、メッセージの消費失敗につながる可能性があります。

RocketMQ 5.0 のメッセージレベルの負荷分散

クライアント側の負荷分散による重複消費と消費遅延の問題を根本的に解決するため、RocketMQ 5.0 ではメッセージレベルの負荷分散メカニズムが導入されました。

同一キュー内のメッセージを複数のコンシューマーで消費できるようになりました。サーバーはメッセージがクライアントによって重複なく、かつ漏れなく消費されることを保証します。

メッセージ粒度の負荷分散メカニズムは、内部の単一メッセージ確認セマンティクスに基づいて実装されています。コンシューマーがメッセージを取得した後、サーバーはそのメッセージをロックし、メッセージが正常に消費されるか消費タイムアウトが発生するまで、他のコンシューマーに対して非表示にします。そのため、複数のコンシューマーが同時に同一キューのメッセージを消費する場合でも、サーバーはメッセージが複数のコンシューマーによって重複消費されないことを保証できます。

4.x クライアントでは、順序消費の実装はキュー割り当てに強く依存していました。RocketMQ 5.0 では、メッセージ次元の負荷分散を基盤として順序消費のセマンティクスも実装されています。異なるコンシューマーが同一メッセージグループ内のメッセージを処理する際、メッセージステータスを厳密な順序でロックし、同一メッセージグループ内のメッセージのシリアル消費を保証します。

上図に示すように、キュー Queue1 には 4 つの順序メッセージがあり、これら 4 つのメッセージは同一メッセージグループ G1 に属し、M1 から M4 の順に格納されています。消費プロセスにおいて、前のメッセージ M1 と M2 がコンシューマー A1 によって処理されている間、消費ステータスがコミットされていない限り、コンシューマー A2 は後続の M3 と M4 を並列に消費できず、前のメッセージの消費ステータスがコミットされるのを待ってから後続のメッセージを消費する必要があります。