コンシューマーとは、トピックをサブスクライブし、ApsaraMQ for RocketMQ からメッセージを受信・処理するクライアントアプリケーションです。コンシューマーはブローカーに接続し、自身のサブスクリプションに基づいてメッセージを取得し、アプリケーションのビジネスロジックへ渡して処理します。
コンシューマーの動作は、以下の 3 つの要素によって決定されます：
コンシューマーグループ：すべてのコンシューマーは、コンシューマーグループに属する必要があります。グループは、配信順序やリトライポリシーなどの共有動作設定を定義し、グループ内のメンバー全体における消費進捗を追跡します。
コンシューマータイプ：ブローカー主導の配信を行うプッシュコンシューマーと、アプリケーション主導の取得（プル）を行うシンプルコンシューマーのいずれかを選択します。詳細については、「コンシューマータイプ」をご参照ください。
クライアント側設定：スレッド数、同時実行数、その他のローカル構成により、ご利用のアプリケーションホスト上でコンシューマーがメッセージをどのように処理するかが制御されます。
仕組み
以下の図は、ApsaraMQ for RocketMQ のドメインモデルにおけるコンシューマーの位置を示しています。
プロデューサーが ApsaraMQ for RocketMQ ブローカーへメッセージを送信します。
ブローカーは、対象トピックで指定されたキューに各メッセージを格納し、到着順序を保持します。
コンシューマーは、自身のサブスクリプションに基づき、ブローカーからメッセージを取得・処理します。
同一のコンシューマーグループに属するコンシューマーは、当該グループへ配信された全メッセージを共有します。
内部属性
- このコンシューマーが所属するコンシューマーグループの名前です。コンシューマーは、所属するコンシューマーグループから配信順序およびリトライポリシーを継承します。
- コンシューマーグループは論理的なリソースです。使用前に、コンソールまたは API を通じて作成してください。命名規則およびその他の制約については、「制限事項」をご参照ください。
- クラスター内における各コンシューマークライアントインスタンスを一意に識別する ID です。ApsaraMQ for RocketMQ SDK がこの値を自動生成します。クライアント ID はログに表示され、トラブルシューティング時に有用ですが、変更することはできません。
- コンシューマーがサブスクライブするトピックおよびフィルター式のリストです。コンシューマーの初期化時にサブスクリプションを指定することで、アプリケーション起動後にではなく、事前にブローカーがトピックの権限および可用性を検証できます。
- 初期化時にサブスクリプションを指定しない場合、またはランタイム中にサブスクライブするトピックが変更される場合、ApsaraMQ for RocketMQ は動的検証を実行します。これにより、エラー検出が遅れる可能性があります。
- ブローカーがプッシュコンシューマーへメッセージを配信する際に呼び出すコールバックインターフェイスです。メッセージリスナーは、コンシューマークライアント側で設定します。
- シンプルコンシューマーはメッセージを明示的に取得するため、リスナーを使用しません。「コンシューマータイプ」をご参照ください。
動作制約
同一のコンシューマーグループに属するすべてのコンシューマーは、以下の設定を同一にする必要があります：
| 設定 | 説明 |
|---|---|
| メッセージ配信順序 | メッセージが逐次的または同時並行で配信されるかを決定します |
| 消費リトライポリシー | 失敗したメッセージのリトライ方法を定義します |
同一グループ内のコンシューマー間で設定が不一致の場合、予測不可能な配信動作が発生します。これらの設定の適用方法は、ブローカーのバージョンによって異なります。詳細については、次のセクションをご参照ください。
バージョン互換性
| 動作 | 5.x ブローカー | 3.x / 4.x ブローカー |
|---|---|---|
| 配信順序およびリトライポリシーの定義場所 | コンシューマーグループ（サーバー側） | 各コンシューマークライアント（クライアント側 API） |
| 開発者の責任 | クライアント側での構成は不要 — コンシューマーはグループから設定を継承します | グループ全体の設定を一貫して保つため、すべてのコンシューマークライアントで配信順序およびリトライポリシーを設定する必要があります |
5.x ブローカーに旧バージョンの SDK を使用して接続する場合、コンシューマーは引き続きクライアント側 API 構成から配信順序およびリトライポリシーを読み取り、コンシューマーグループからの設定は無視されます。
ベストプラクティス
プロセスごとに 1 つのコンシューマーを 1 つのグループに対して初期化する
ApsaraMQ for RocketMQ のコンシューマーは、同時実行およびマルチスレッドアクセスをサポートするノンブロッキングプロトコルを使用します。単一のコンシューマーインスタンスで高スループットを処理できます。ほとんどのケースでは、各プロセス内で 1 つのコンシューマーグループにつき 1 つのコンシューマーを初期化してください。同一プロセス内で同一構成の複数のコンシューマーインスタンスを作成すると、パフォーマンス向上には寄与せず、リソースを無駄に消費します。
コンシューマーの作成・破棄ではなく再利用を行う
コンシューマーインスタンスはデータベース接続プールと同様に扱い、アプリケーションライフサイクル全体を通して一度作成し、再利用してください。メッセージのバッチごとに新しいコンシューマーを作成・破棄すると、ブローカー上に過剰な短時間接続が発生し、システム負荷を不必要に増加させます。
正しい例 — 一度作成し、ループ内で消費、終了時にシャットダウン：
Consumer c = ConsumerBuilder.build();
for (int i = 0; i < n; i++) {
Message m = c.receive();
// メッセージを処理
}
c.shutdown();誤った例 — ループ内で作成・破棄：
for (int i = 0; i < n; i++) {
Consumer c = ConsumerBuilder.build();
Message m = c.receive();
// メッセージを処理
c.shutdown();
}次のステップ
コンシューマーグループ：コンシューマーグループが共有配信設定を管理し、消費進捗を追跡する方法について学びます。
コンシューマータイプ：プッシュコンシューマーとシンプルコンシューマーを比較し、ご利用のユースケースに最適なタイプを選択します。
制限事項：コンシューマーグループの命名規則、クォータ、その他の制約事項を確認します。