如何排查RocketMQ订阅关系不一致的问题？

商业版JAVA SDK

maven坐标样例：

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

示例代码：

Properties properties = new Properties();
// 您在消息队列RocketMQ版控制台创建的Group ID。
properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-xxxxx");
/*
* 一些属性配置
*/
Consumer consumer = ONSFactory.createConsumer(properties);
//订阅多个Tag。
consumer.subscribe("TopicTestMQ1", "TagA||TagB", new MessageListener() {...});
//订阅多个Topic,同一个consumer实例继续调用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>

示例代码：

// 您在消息队列RocketMQ版控制台创建的Group ID。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP_ID, getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);
/*
* 一些属性配置
*/
//订阅多个Tag
consumer.subscribe("java-test", "TagA||TagB");
//订阅第二个Topic，需要注意consumer只能注册一个消费逻辑，即多个订阅的消费逻辑都是一样的
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<>();
//订阅多个Tag
sub.put("Topic1", new FilterExpression("TagA||TagB", FilterExpressionType.TAG));
//订阅第二个topic，需要注意consumer只能注册一个消费逻辑，即多个订阅的消费逻辑都是一样的
sub.put("Topic2", new FilterExpression("TagA||TagB", FilterExpressionType.TAG));
PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                //设置消费者分组，消息队列RocketMQ版控制台创建的Group ID。
                .setConsumerGroup(consumerGroup)
                //设置预绑定的订阅关系。
                .setSubscriptionExpressions(sub)
                //设置消费监听器。
                .setMessageListener(messageView -> {...})
                .build();

SpringBoot框架（使用RocketMQMessageListener注解）

若使用RocketMQMessageListener注解来启动消费者，需要留意每个注解都会启动一个独立的consumer实例，若代码中使用了多次注解，且每处注解订阅的Topic不同，就会导致订阅关系不一致。

比如，下面这段代码实现了两次RocketMQListener接口，在同一个Group中分别订阅了TopicA和TopicB，实际运行时客户端会创建两个consumer实例，最终导致在同一个消费者组内有两个消费者分别订阅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{
            // 订阅多个tag
            defaultMQPushConsumer.subscribe("TopicA", "TagA||TagB");
            // 订阅第二个topic
            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() {
        //参考商业版以及社区版SDK多订阅的方式在此处注入对应的ConsumerBean
    }
}