ApsaraMQ for RocketMQ:コンシューマーの種類

タイムアウト時の動作

処理ロジックがブロックされ、許容時間内にメッセージ処理が完了しない場合、SDK は強制的に失敗結果を送信し、リトライポリシーに従ってメッセージを処理します。

PushConsumer の例

// PushConsumer を使用して通常メッセージを消費します
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        // コンシューマーグループを設定します
        .setConsumerGroup("<your-consumer-group>")
        // アクセスエンドポイントを設定します
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // サブスクリプションをバインドします
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // メッセージリスナーを登録します
        .setMessageListener(new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                // メッセージを処理し、結果を返します
                return ConsumeResult.SUCCESS;
            }
        })
        .build();

以下のプレースホルダーを実際の値に置き換えてください。

LitePushConsumer の例

// LitePushConsumer を使用して通常メッセージを消費します
ClientServiceProvider provider = ClientServiceProvider.loadService();

LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
        // アクセスエンドポイントを設定します
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // トピックを設定します
        .bindTopic("<your-topic>")
        // コンシューマーグループを設定します
        .setConsumerGroup("<your-consumer-group>")
        // メッセージリスナーを登録します
        .setMessageListener(messageView -> {
            // メッセージを処理し、結果を返します
            return ConsumeResult.SUCCESS;
        })
        .build();

// Lite トピックをサブスクライブします
litePushConsumer.subscribeLite("<your-lite-topic-1>");
litePushConsumer.subscribeLite("<your-lite-topic-2>");

SimpleConsumer の例

// SimpleConsumer を使用して通常メッセージを消費します
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        // コンシューマーグループを設定します
        .setConsumerGroup("<your-consumer-group>")
        // アクセスエンドポイントを設定します
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // サブスクリプションをバインドします
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        .build();

try {
    // 最大 10 件のメッセージをプルし、最大 30 秒待機します
    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) {
    // スロットリングなどの障害を処理し、receive 呼び出しを再試行します
    e.printStackTrace();
}