ApsaraMQ for RocketMQ:Java SDK を使用したメッセージの送受信

プッシュコンシューマー

PushConsumerExample.java ファイルを作成して実行します：

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // インターネット経由でアクセスされるサーバーレスインスタンスの場合は、コメントアウトを解除します：
                // .setNamespace("<your-instance-id>")
                // インターネットアクセス、または認証が必要なサーバーレス VPC の場合は、コメントアウトを解除します：
                // .setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"))
                .build();

        // トピック内のすべてのメッセージをサブスクライブします (tag = "*")
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    System.out.println("Consume Message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();

        // コンシューマーを実行し続けます
        Thread.sleep(Long.MAX_VALUE);

        // コンシューマーを正常にシャットダウンするには、次を呼び出します：
        // pushConsumer.close();
    }
}

実行に時間のかかるビジネスロジックには、代わりにシンプルコンシューマー (プルモード) を使用してください。シンプルコンシューマーを使用すると、一度にプルするメッセージの数を正確に制御でき、現在のバッチを処理して確認応答した後にのみ次のバッチをプルします。InvisibleDuration (受信したメッセージが他のコンシューマーから見えなくなる期間、10 秒から 12 時間) を設定して確認応答のタイムアウトを制御し、必要に応じて動的に延長することができます。

シンプルコンシューマー

SimpleConsumerExample.java ファイルを作成して実行します：

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    private SimpleConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // インターネット経由でアクセスされるサーバーレスインスタンスの場合は、コメントアウトを解除します：
                // .setNamespace("<your-instance-id>")
                // インターネットアクセス、または認証が必要なサーバーレス VPC の場合は、コメントアウトを解除します：
                // .setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"))
                .build();

        // ロングポーリングのタイムアウト
        Duration awaitDuration = Duration.ofSeconds(10);
        // トピック内のすべてのメッセージをサブスクライブします (tag = "*")
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(awaitDuration)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();

        // 1回のプルあたりの最大メッセージ数
        int maxMessageNum = 16;
        // 受信後、この期間メッセージは他のコンシューマーから見えなくなります
        Duration invisibleDuration = Duration.ofSeconds(10);

        // ループでメッセージをポーリングします。リアルタイムで消費するには、複数のスレッドを使用してください。
        while (true) {
            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
            messages.forEach(messageView -> {
                System.out.println("Received message: " + messageView);
            });
            for (MessageView message : messages) {
                final MessageId messageId = message.getMessageId();
                try {
                    // 各メッセージを確認応答して、消費結果をブローカーにコミットします
                    consumer.ack(message);
                    System.out.println("Message is acknowledged successfully, messageId= " + messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
        // コンシューマーを正常にシャットダウンするには、次を呼び出します：
        // consumer.close();
    }
}