:RocketMQ におけるサブスクリプション関係の不整合のトラブルシューティング方法

商用 Java SDK

Maven 座標の例：

<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>x.x.x.Final</version>

サンプルコード：

Properties properties = new Properties();
// ApsaraMQ for RocketMQ コンソールで作成した Group ID。
properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-xxxxx");
/*
* いくつかのプロパティ設定
*/
Consumer consumer = ONSFactory.createConsumer(properties);
// 複数のタグをサブスクライブします。
consumer.subscribe("TopicTestMQ1", "TagA||TagB", new MessageListener() {...});
// 複数のトピックをサブスクライブします。同じコンシューマーインスタンスで再度 subscribe を呼び出してサブスクリプションを追加します。
consumer.subscribe("TopicTestMQ2", "TagA||TagB", new MessageListener() {...});

コミュニティ Java SDK (artifactId: rocketmq-client)

Maven 座標の例：

<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>x.x.x</version>

サンプルコード：

// ApsaraMQ for RocketMQ コンソールで作成した Group ID。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP_ID, getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);
/*
* いくつかのプロパティ設定
*/
// 複数のタグをサブスクライブします。
consumer.subscribe("java-test", "TagA||TagB");
// 2 つ目のトピックをサブスクライブします。コンシューマーは 1 つの消費ロジックしか登録できないことに注意してください。これは、すべてのサブスクリプションに同じロジックが適用されることを意味します。
consumer.subscribe("cxtest", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {...});

コミュニティ Java SDK (artifactId: rocketmq-client-java)

Maven 座標の例：

<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client-java</artifactId>
<version>5.0.8</version>

サンプルコード：

final ClientServiceProvider provider = ClientServiceProvider.loadService();
/*
* いくつかの設定
*/
HashMap<String, FilterExpression> sub = new HashMap<>();
// 複数のタグをサブスクライブします。
sub.put("Topic1", new FilterExpression("TagA||TagB", FilterExpressionType.TAG));
// 2 つ目のトピックをサブスクライブします。コンシューマーは 1 つの消費ロジックしか登録できないことに注意してください。これは、すべてのサブスクリプションに同じロジックが適用されることを意味します。
sub.put("Topic2", new FilterExpression("TagA||TagB", FilterExpressionType.TAG));
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // コンシューマーグループを設定します。ApsaraMQ for RocketMQ コンソールで作成した Group ID を使用します。
                .setConsumerGroup(consumerGroup)
                // 事前にアタッチされたサブスクリプション関係を設定します。
                .setSubscriptionExpressions(sub)
                // メッセージリスナーを設定します。
                .setMessageListener(messageView -> {...})
                .build();

SpringBoot フレームワーク (`RocketMQMessageListener` アノテーションを使用)

RocketMQMessageListener アノテーションを使用してコンシューマーを起動する場合、各アノテーションが個別のコンシューマーインスタンスを起動することに注意してください。コード内で複数のアノテーションを使用し、それぞれが異なるトピックをサブスクライブすると、サブスクリプション関係の不整合が発生します。

例えば、以下のコードは `RocketMQListener` インターフェイスを 2 回実装しており、同じグループで `TopicA` と `TopicB` をサブスクライブします。実行時に、クライアントは 2 つのコンシューマーインスタンスを作成します。これにより、同じコンシューマーグループ内の 2 つのコンシューマーがそれぞれ TopicA と TopicB をサブスクライブすることになり、サブスクリプション関係の不整合が発生します。

// ビジネスロジック 1: TestMessageListener1.java
@Component
@RocketMQMessageListener(consumerGroup="GID_A", topic = "TopicA")
public class TestMessageListener1 implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        /*
        * いくつかの設定
        */
    }
}
// ビジネスロジック 2: TestMessageListener2.java
@Component
@RocketMQMessageListener(consumerGroup="GID_A", topic = "TopicB")
public class TestMessageListener2 implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        /*
        * いくつかの設定
        */
    }
}

複数のサブスクリプションを実装するには、RocketMQPushConsumerLifecycleListener インターフェイスの prepareStart メソッドを再書き込みします。以下のコードに例を示します。

@Component
@RocketMQMessageListener(consumerGroup = "GID_A", topic = "")
public class TestMessageListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(MessageExt messageExt) {}
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        try{
            // 複数のタグをサブスクライブします。
            defaultMQPushConsumer.subscribe("TopicA", "TagA||TagB");
            // 2 つ目のトピックをサブスクライブします。
            defaultMQPushConsumer.subscribe("TopicB", "*");
            // 消費ロジックを登録します。
            defaultMQPushConsumer.registerMessageListener((List<MessageExt> messages, ConsumeConcurrentlyContext context) -> {......});
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

SpringBoot フレームワーク (`Bean` アノテーションを使用)

このメソッドは ConsumerBean を手動でインジェクションします。複数のサブスクリプションを作成するには、ネイティブ SDK のメソッドを使用する必要があります。

@Configuration
public class ConsumerClient {
  
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean buildConsumer() {
        // ここで対応する ConsumerBean をインジェクションします。商用およびコミュニティ SDK の複数サブスクリプションメソッドをご参照ください。
    }
}