Spring Boot から RocketMQ 5.0 への接続 - ApsaraMQ for RocketMQ

このトピックでは、rocketmq-spring-boot-starter と rocketmq-v5-client-spring-boot-starter を使用して、Spring Boot 環境で ApsaraMQ for RocketMQ 5.0 インスタンスに素早く接続する方法について説明します。

背景情報

Spring Boot スターターパッケージは、プロデューサーとコンシューマーを作成するロジックをカプセル化しています。Remoting プロトコルには rocketmq-client を、gRPC プロトコルには rocketmq-client-java を使用します。

他の SDK の詳細については、「https://www.alibabacloud.com/help/en/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-5-x-series/developer-reference/overview-8」をご参照ください。

rocketmq-spring-boot-starter

依存関係のインポート

<dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-spring-boot-starter</artifactId>
      <version>{REPLACE_WITH_ACTUAL_VERSION}</version>
 </dependency>

説明

このスターターは rocketmq-client に依存しています。
バージョン番号については、Maven リポジトリをご参照ください。

プロジェクト設定ファイル

# メッセージトレース機能
rocketmq.access-channel=CLOUD
rocketmq.name-server={YOUR_ENDPOINT}
rocketmq.consumer.access-key={YOUR_USERNAME}
rocketmq.consumer.secret-key={YOUR_PASSWORD}
rocketmq.producer.access-key={YOUR_USERNAME}
rocketmq.producer.secret-key={YOUR_PASSWORD}
#rocketmq.producer.namespaceV2={YOUR_INSTANCE_ID}
rocketmq.producer.group=test

説明

前述の変数を実際の値に置き換えてください。中括弧 {} は含めないでください。

エンドポイントに http:// プレフィックスを追加しないでください。
パブリックネットワーク経由でサーバーレスインスタンスに接続する場合は、`namespaceV2` パラメーターを設定する必要があります。

送信コード例

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

Message<String> msg = MessageBuilder.withPayload("Hello,RocketMQ").build();
// 通常メッセージを送信します。Topic が作成されていることを確認してください。
SendResult sendResult = rocketMQTemplate.syncSend("TEST_TOPIC:mytag", msg); // TEST_TOPIC は Topic 名、mytag はタグ名です。
// 遅延メッセージを送信します。Topic の作成時に、メッセージタイプを「予約/遅延メッセージ」に設定してください。
SendResult delaySendResult = rocketMQTemplate.syncSendDelayTimeMills("delay:mytag", msg, 6000);

消費コード例

@Component
@RocketMQMessageListener(topic = "TEST_TOPIC",selectorExpression = "*", consumerGroup = "GID_test",enableMsgTrace = true,messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.CONCURRENTLY,accessChannel = "CLOUD")
public class MyMQListener implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        System.out.println("msg id is " + message.getMsgId() + " , msg body is " + new String(message.getBody()));
    }
}

説明

サーバーレスインスタンスにパブリックネットワーク経由で接続する場合、RocketMQMessageListener のプロパティに namespaceV2={YOUR_INSTANCE_ID} も指定する必要があります。

rocketmq-v5-client-spring-boot-starter

依存関係のインポート

<dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-v5-client-spring-boot-starter</artifactId>
      <version>{REPLACE_WITH_ACTUAL_VERSION}</version>
 </dependency>

説明

このスターターは rocketmq-client-java に依存します。
バージョン番号については、Maven リポジトリをご参照ください。

プロジェクト設定ファイル

rocketmq.access-channel=CLOUD
rocketmq.push-consumer.endpoints={YOUR_ENDPOINT}
rocketmq.push-consumer.access-key={YOUR_USERNAME}
rocketmq.push-consumer.secret-key={YOUR_PASSWORD}
rocketmq.producer.access-key={YOUR_USERNAME}
rocketmq.producer.secret-key={YOUR_PASSWORD}
rocketmq.producer.namespace={YOUR_INSTANCE_ID}
rocketmq.producer.endpoints={YOUR_ENDPOINT}

送信コード例

    @Autowired
    private RocketMQClientTemplate rocketMQClientTemplate;

User user = new User();
user.setName(body);
user.setAge(18);
SendReceipt springbootv5 = rocketMQClientTemplate.syncSendNormalMessage("TEST_TOPIC:mytag", user); // TEST_TOPIC は Topic 名、mytag はタグ名です。

消費コード例

@Component
@RocketMQMessageListener(topic = "data", consumerGroup = "GID_test", namespace = "{YOUR_INSTANCE_ID}", tag = "*")
public class MyConsumer implements RocketMQListener {
    @Override
    public ConsumeResult consume(MessageView messageView) {
        System.out.println(messageView.getMessageId() + ": body is " + StandardCharsets.UTF_8.decode(messageView.getBody()));
        return ConsumeResult.SUCCESS;
    }
}

説明

サーバーレスインスタンスの場合、`namespace` プロパティをご利用のインスタンス ID に設定します。
メッセージ本文は ByteBuffer です。トランスコードする必要があります。この例では StandardCharsets.UTF_8.decode(messageView.getBody()) を使用します。
メッセージトレースを有効にするには、enableMsgTrace = true プロパティを追加します。

必要な情報の取得

エンドポイントの取得

ご利用の ApsaraMQ for RocketMQ インスタンスの詳細ページに移動します。 [基本情報] タブの [TCP プロトコルエンドポイント] セクションで、[エンドポイントとネットワーク情報] を取得します。 [VPC] と [パブリックネットワーク] のアクセスタイプの違いにご注意ください。

ユーザー名とパスワードの取得

ご利用の ApsaraMQ for RocketMQ インスタンスの詳細ページに移動します。左側のナビゲーションウィンドウで、[Resource Access Management] をクリックします。 [認証情報] タブを選択します。 [ユーザー名] と [パスワード] を記録します。