ApsaraMQ for RocketMQ:コンシューマータイプ

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 ConsumerGroup")
                // エンドポイントを設定します。
                .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())
        // Topic を設定します。
        .bindTopic("Your Topic")
        // コンシューマーグループを設定します。
        .setConsumerGroup("Your ConsumerGroup")
        // メッセージリスナーを設定します。
        .setMessageListener(messageView -> {
            // メッセージを消費し、処理結果を返します。
            return ConsumeResult.SUCCESS;
        })
        .build();

// ターゲットの Lite Topic をサブスクライブします。
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 ConsumerGroup")
                // エンドポイントを設定します。
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                // 事前にバインドされたサブスクリプション関係を設定します。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        List<MessageView> messageViewList = null;
        try {
            // SimpleConsumer は、積極的にメッセージを取得して処理する必要があります。
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // 処理が完了したら、積極的に ack() を呼び出して消費結果をコミットします。
                try {
                    simpleConsumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            // システムのスロットリングなどの理由でプルが失敗した場合は、リクエストを再開してメッセージを取得する必要があります。
            e.printStackTrace();
        }

SimpleConsumer は、主に次の API 操作を含みます。