ApsaraMQ for RocketMQ:コンシューマーの負荷分散

仕組み

ブローカーは、メッセージが属するキューに関係なく、トピックからの個々のメッセージをコンシューマーグループ内のすべてのコンシューマーに均等に分散します。複数のコンシューマーが同じキューのメッセージを同時に処理できます。

この例では、コンシューマーグループ A には A1、A2、A3 の 3 つのコンシューマーがあります。3 つすべてが Queue1 からメッセージを消費します。ブローカーは各メッセージを一度に 1 つのコンシューマーに割り当てます。

コンシューマーがメッセージを受信すると、ブローカーはそのメッセージをロックし、他のコンシューマーから見えないようにします。メッセージは、コンシューマーが確認応答するか、ロックがタイムアウトするまでロックされたままになります。これにより、通常の条件下での重複処理が防止されます。

順序付きメッセージの処理

順序付きメッセージの場合、ApsaraMQ for RocketMQ は、同じメッセージグループ内のメッセージがブローカーに保存された順序どおりに処理されることを保証します。

Queue1 のメッセージグループ G1 にある 4 つの順序付きメッセージ (M1 から M4) を考えます。コンシューマー A1 が M1 と M2 を処理している場合、コンシューマー A2 は、A1 が M1 と M2 の消費ステータスを送信するまで、M3 または M4 の処理を開始できません。ブローカーは、順序を維持するために厳密なシーケンシャルロックを強制します。

キューベースの負荷分散に対するメリット

利用シーン

メッセージベースの負荷分散は、注文処理、通知ディスパッチ、リアルタイムイベント処理など、各メッセージが独立して処理されるほとんどのオンラインイベント処理ワークロードに適しています。

同じキューからのメッセージをバッチ処理する必要があるストリーム処理や集約ワークロードの場合は、代わりにキューベースの負荷分散を使用してください。

適用範囲

メッセージベースの負荷分散は、ブローカーバージョン 5.x の Push コンシューマーおよび Simple コンシューマータイプで利用できる唯一のポリシーです。デフォルトで有効になっており、追加の構成は必要ありません。

例

Push コンシューマーと Simple コンシューマータイプは、メッセージベースの負荷分散を自動的に使用します。以下の Java の例は、両方のコンシューマータイプが負荷分散の構成なしでメッセージを処理する方法を示しています。

メッセージリスナーを持つ Push コンシューマー：

// Push コンシューマー：メッセージを処理するためにメッセージリスナーを実装します。
// 負荷分散はブローカーによって自動的に処理されます。
MessageListener messageListener = new MessageListener() {
    @Override
    public ConsumeResult consume(MessageView messageView) {
        System.out.println(messageView);
        // 消費結果を返します。
        return ConsumeResult.SUCCESS;
    }
};

手動での確認応答を行う Simple コンシューマー：

// Simple コンシューマー：メッセージをプルし、処理し、それぞれを確認応答します。
// 負荷分散はブローカーによって自動的に処理されます。
try {
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        try {
            // 処理後にメッセージを確認応答します。
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // スロットリングやその他の問題でプルが失敗した場合は、リクエストをリトライします。
    e.printStackTrace();
}

仕組み

ブローカーは、トピック内の各キューをコンシューマーグループ内のただ 1 つのコンシューマーに割り当てます。各コンシューマーは、割り当てられたキューからのすべてのメッセージを処理します。

この例では、トピックには 3 つのキュー (Queue1、Queue2、Queue3) があり、コンシューマーグループには 2 つのコンシューマーがあります。各キューは 1 つのコンシューマーに割り当てられるため、コンシューマー A2 は 2 つのキューを取得し、A1 は 1 つのキューを取得します。コンシューマーよりもキューが少ない場合、一部のコンシューマーはキューを受信せず、アイドル状態のままになります。

各コンシューマーは、割り当てられたキューからメッセージをプルし、消費オフセットを送信し、オフセットを永続化するという処理シーケンスに従います。コンシューマーがメッセージをプルする際に消費ステータスがキューに返されないため、重複処理を防ぐには、各キューを単一のコンシューマーに排他的に割り当てる必要があります。

メッセージベースの負荷分散に対するメリット

利用シーン

キューベースの負荷分散は、同じソースからのメッセージをバッチ処理または集約する必要があるストリームコンピューティングやデータ集約アプリケーションに最適です。たとえば、コンシューマーがタイムウィンドウにわたって単一のキューからメトリックを収集し、実行平均を計算するような場合です。

適用範囲

キューベースの負荷分散は、Pull コンシューマー、デフォルトの Push コンシューマー、デフォルトの Pull コンシューマー、Lite Pull コンシューマータイプなど、ブローカーバージョン 3.x および 4.x のコンシューマーで利用できる唯一のポリシーです。ブローカーバージョン 5.x では、Pull コンシューマータイプはデフォルトでキューベースの負荷分散を引き続き使用します。

これらのコンシューマータイプでは、キューベースの負荷分散が自動的に有効になるため、追加の構成は必要ありません。

例

サンプルコードについては、Apache RocketMQ コードライブラリの LitePullConsumerAssign.java をご参照ください。

べき等なメッセージ処理の実装

どちらの負荷分散ポリシーも、コンシューマーが追加、削除されたり、ブローカーがスケーリングされたりすると、一時的なリバランスをトリガーします。リバランス中、少数のメッセージが複数回配信される可能性があります。これに対処するには、重複排除を実装して、メッセージ消費ロジックのべき等性を確保してください。