# 实现订阅关系一致 #### 方案概述 订阅关系一致指的是同一个消费组下所有消费者所订阅的Topic、Tag必须完全一致。如果订阅关系不一致，消息消费的逻辑就会混乱，甚至导致消息丢失。 **消费原理** RocketMQ为每个Topic划分了消息队列（Queue），队列数越大消费的并发度越大。一个消费组表示一个消费者群组，在分布式应用场景下，同一个消费组中的多个消费者共同完成Topic所有Queue的消费。Queue的分配以消费组为单位，会均匀分配给消费组下的消费者，而不会在意该消费者是否订阅了当前Topic。一个消费者只会分配到Topic中的某几个Queue，而一个Queue只会分配给一个消费者。 **正确的订阅关系** 在分布式应用场景下，一个消费组中所有的消费者拥有一个相同的消费组ID，他们需订阅相同的Topic和Tag，保持订阅关系一致，才能保证消息消费逻辑正确，消息不丢失。 - 同一个消费组的消费者必须订阅同一个Topic。例如，消费组Group1中有消费者A和消费者B，消费者A订阅了Topic A和Topic B，则消费者B也必须订阅Topic A和Topic B，不能只订阅Topic A或只订阅Topic B，或者额外订阅Topic C。 - 同一个消费组的消费者订阅同一个Topic下的Tag必须一致，包括Tag的数量和顺序。例如，消费组Group2中有消费者A和消费者B，消费者A订阅Topic A且Tag为Tag A\|\|Tag B，则消费者B订阅Topic A时Tag也必须为Tag A\|\|Tag B，不能只订阅Tag A或只订阅Tag B或订阅Tag B\|\|Tag A。 图1正确的订阅关系一致设置 ![](https://support.huaweicloud.com/bestpractice-hrm/zh-cn_image_0000001658265353.png "点击放大") 订阅关系一致性保证了同一个消费组中消费消息的正常运行，避免消息逻辑混乱和消息丢失。在实际使用中，生产者端要做好消息的分类，便于消费者可以使用Tag进行消息的准确订阅。而在消费者端，则要保证订阅关系的一致性。 **错误的订阅关系** - 同一消费组下的消费者订阅了不同的Topic 例如，消费组Group1下有消费者A和消费者B，消费者A订阅了Topic A，消费者B订阅了Topic B。当生产者向Topic A发送消息时，消息会按Queue均匀发送给消费者A和消费者B。由于消费者B没有订阅Topic A的消息，会把Topic A消息过滤掉（即[图2]中Topic A的Queue2中的消息会被消费者B过滤），从而导致部分Topic A消息未被消费。 图2错误的Topic订阅 ![](https://support.huaweicloud.com/bestpractice-hrm/zh-cn_image_0000001608867628.png) - 同一消费组下的消费者订阅了相同Topic下不同的Tag 例如，消费组Group1下有消费者A和消费者B，消费者A订阅了Topic A的Tag A，消费者B订阅了Topic A的Tag B。当生产者A向Topic A发送Tag A的消息时，Tag A消息会按Queue均匀发送给消费者A和消费者B。由于消费者B没有订阅Tag A的消息，会把Tag A消息过滤掉（即[图3]中Queue2中的Tag A消息会被消费者B过滤），从而导致部分Tag A消息未被消费。 图3错误的Tag订阅 ![](https://support.huaweicloud.com/bestpractice-hrm/zh-cn_image_0000002008601125.png "点击放大") #### 特性讲解 #### 实施方法 - **订阅一个Topic且订阅一个Tag** 同一消费组Group1中的消费者Consumer1、Consumer2和Consumer3都订阅了Topic_A，且都订阅了Topic_A的Tag_A，符合订阅关系一致原则。Consumer1、Consumer2、Consumer3的订阅关系一致，即Consumer1、Consumer2、Consumer3订阅消息的代码必须完全一致，代码示例如下： ``` DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Group1"); consumer.subscribe("Topic_A", "Tag_A"); ``` - **订阅一个Topic且订阅多个Tag** 同一消费组Group1中的消费者Consumer1、Consumer2和Consumer3都订阅了Topic_A，且都订阅了Topic_A的Tag_A和Tag_B（即订阅Topic_A中所有Tag为Tag_A或Tag_B的消息），顺序都是Tag_A\|\|Tag_B，符合订阅关系一致性原则。Consumer1、Consumer2、Consumer3的订阅关系一致，即Consumer1、Consumer2、Consumer3订阅消息的代码必须完全一致，代码示例如下： ``` DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Group1"); consumer.subscribe("Topic_A", "Tag_A||Tag_B"); ``` - **订阅多个Topic且订阅多个Tag** 同一消费组Group1中的消费者Consumer1、Consumer2和Consumer3都订阅了Topic_A和Topic_B，且订阅Topic_A都未指定Tag（即订阅Topic_A中的所有消息），订阅Topic_B的Tag都是Tag_A和Tag_B（即订阅Topic_B中所有Tag为Tag_A或Tag_B的消息），顺序都是Tag_A\|\|Tag_B，符合订阅关系一致性原则。Consumer1、Consumer2、Consumer3的订阅关系一致，即Consumer1、Consumer2、Consumer3订阅消息的代码必须完全一致，代码示例如下： ``` DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Group1"); consumer.subscribe("Topic_A", "*"); consumer.subscribe("Topic_B", "Tag_A||Tag_B"); ``` #### 相关操作 查看订阅关系是否一致的方法请参见[查看RocketMQ消费组和Topic的订阅关系](https://support.huaweicloud.com/usermanual-hrm/hrm-ug-087.html)。