ApsaraMQ for RocketMQ:バッチ消費

仕組み

プッシュコンシューマーは、2 つの段階でバッチ消費を処理します：

1. プルとキャッシュ：メッセージプルスレッドは、ロングポーリングを使用して ApsaraMQ for RocketMQ からメッセージを取得し、ローカルにキャッシュします。

2. ディスパッチ：キャッシュされたメッセージがバッチサイズのしきい値または待機時間のしきい値に達すると (どちらか早い方)、プッシュコンシューマーはバッチをコンシューマースレッドに送信して処理します。

利用シーン

バッチ消費は、ダウンストリームシステムがバッチ処理からメリットを得られる場合に最も効果的です。並列処理の向上のみが目的の場合は、まずコンシューマーインスタンスの追加やスレッドプールサイズの調整など、より簡単な代替手段を検討してください。

• 一括インデックス作成：アップストリームの注文システムがログメッセージを公開し、ダウンストリームの Elasticsearch クラスターがそれをインデックス化します。各メッセージは 1 つの RPC リクエスト (約 10 ms) をトリガーします。10 件のメッセージを個別に処理すると 100 ms かかりますが、それらを単一の一括インデックス作成呼び出しにまとめることで、合計時間は約 10 ms に短縮されます。

• データベースへの一括挿入：アプリケーションが高い更新頻度で一度に 1 つずつレコードをデータベースに挿入すると、高い負荷が発生します。挿入ごとに 10 レコードをバッチ処理し、5 秒ごとにフラッシュすることで、接続のオーバーヘッドと書き込み増幅が削減されます。

制限事項

• バッチ消費は TCP 経由でのみサポートされます。TCP クライアント SDK for Java の商用版、バージョン 1.8.7.3.Final 以降を使用してください。リリースノートとダウンロード手順については、「リリースノート」をご参照ください。

• 最大バッチサイズ：1,024 メッセージ。

• バッチ間の最大待機時間：450 秒。

パラメーター

バッチがいつディスパッチされるかは、2 つのパラメーターでコントロールします。ディスパッチは、いずれかの条件が最初に満たされたときに発生します。

サンプルコード

ONSFactory.createBatchConsumer() に渡される Properties を通じてバッチ消費を設定します。BatchMessageListener コールバックは、最大で ConsumeMessageBatchMaxSize 個のメッセージを含む List<Message> を受け取ります。

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import java.util.List;
import java.util.Properties;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.tcp.example.MqConfig;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        consumerProperties.setProperty(PropertyKeyConst.GROUP_ID, MqConfig.GROUP_ID);
        consumerProperties.setProperty(PropertyKeyConst.AccessKey, MqConfig.ACCESS_KEY);
        consumerProperties.setProperty(PropertyKeyConst.SecretKey, MqConfig.SECRET_KEY);
        consumerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MqConfig.NAMESRV_ADDR);

        // バッチあたりの最大メッセージ数を設定します。
        // デフォルト値：32。有効値：1～1024。
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // バッチ間の最大待機時間 (秒) を設定します。
        // デフォルト値：0。有効値：0～450。
        consumerProperties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, String.valueOf(10));

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // メッセージをバッチで処理します。
                return Action.CommitMessage;
            }
        });
        // BatchConsumer を開始します。
        batchConsumer.start();
        System.out.println("Consumer start success.") ;

        // プロセスが終了しないように、一定期間待機します。
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

ベストプラクティス

参考文献

• TCP クライアント SDK for Java の商用版