【RocketMQ工作原理】订阅关系的一致性

爱被打了一巴掌 2022-08-28 15:51 231阅读 0赞

订阅关系的一致性指的是，同一个消费者组（Group ID相同）下所有Consumer实例所订阅的Topic与  
Tag及对消息的处理逻辑必须完全一致。否则，消息消费的逻辑就会混乱，甚至导致消息丢失。

## 1 正确订阅关系 ##

多个消费者组订阅了多个Topic，并且每个消费者组里的多个消费者实例的订阅关系保持了一致。

![image-20210910230933439][]

## 2 错误订阅关系 ##

一个消费者组订阅了多个Topic，但是该消费者组里的多个Consumer实例的订阅关系并没有保持一致。

![image-20210910230959011][]

### 订阅了不同Topic ###

该例中的错误在于，同一个消费者组中的两个Consumer实例订阅了不同的Topic。

*  Consumer实例1-1：（订阅了topic为jodie\_test\_A，tag为所有的消息）

Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_jodie_test_1");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("jodie_test_A", "*", new MessageListener() {
    	public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
    	}
    });

*  Consumer实例1-2：（订阅了topic为jodie\_test\_B，tag为所有的消息）

Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_jodie_test_1");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("jodie_test_B", "*", new MessageListener() {
    	public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

### 订阅了不同Tag ###

该例中的错误在于，同一个消费者组中的两个Consumer订阅了相同Topic的不同Tag。

*  Consumer实例2-1：（订阅了topic为jodie\_test\_A，tag为TagA的消息）

Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_jodie_test_2");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("jodie_test_A", "TagA", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

*  Consumer实例2-2：（订阅了topic为jodie\_test\_A，tag为所有的消息）

Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_jodie_test_2");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("jodie_test_A", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

### 订阅了不同数量的Topic ###

该例中的错误在于，同一个消费者组中的两个Consumer订阅了不同数量的Topic。

*  Consumer实例3-1：（该Consumer订阅了两个Topic）

Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_jodie_test_3");
    Consumer consumer = ONSFactory.createConsumer(properties);
    consumer.subscribe("jodie_test_A", "TagA", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });
    consumer.subscribe("jodie_test_B", "TagB", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            System.out.println(message.getMsgID());
            return Action.CommitMessage;
        }
    });

# RocketMq ConsumerGroup Tag对消费的影响 #

# 官方对consumer的定义如下： #

> Similar to previously mentioned producer group, consumers of the exactly same role are grouped together and named Consumer Group.  
> Consumer Group is a great concept with which achieving goals of load-balance and fault-tolerance, in terms of message consuming, is super easy.  
> Warning: consumer instances of a consumer group must have exactly the same topic subscription(s).

大意是消费者准确按照相同角色来分组，分组的目的是负载均衡和失败转移，并且警告同个分组中的消费者一定要订阅相同的topic。  
在代码中看到consumerGroup变量定义注释如下，大意差不多。

> Consumers of the same role is required to have exactly same subscriptions and consumerGroup to correctly achieve load balance. It’s required and needs to be globally unique.

通过上面的定义，可以准确知道同一组的consumer一定要订阅相同的topic，那么问题来了，订阅的时候除了topic还有tags，这个tag会有影响吗？没有找到相关资料，我就自己做了测试。

# 测试目标：测试在相同的消费组中的消费者，订阅相同的topic时，tag不同会不会影响消费和负载均衡。 #

## 测试计划： ##

*  1.创建生产者发送100条消息；
 *  2.创建消费者

<table> 
 <thead> 
  <tr> 
   <th>instanceName</th> 
   <th>groupName</th> 
   <th>Topic</th> 
   <th>Tag</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>A</td> 
   <td>GroupA</td> 
   <td>TopicA</td> 
   <td>TagA</td> 
  </tr> 
  <tr> 
   <td>B</td> 
   <td>GroupA</td> 
   <td>TopicA</td> 
   <td>TagB</td> 
  </tr> 
 </tbody> 
</table>

*  3.观察消息消费情况和队列分配情况。

## 发消息 ##

public class SimpleProducer {
        public static void sendSync() throws Exception {
            ClientConfig clientConfig=new ClientConfig();
            clientConfig.setNamesrvAddr("localhost:9876");
            MQClientInstance clientInstance=MQClientManager.getInstance().getAndCreateMQClientInstance(clientConfig);
            DefaultMQProducer producer = clientInstance.getDefaultMQProducer();
            producer.setProducerGroup("GroupA");
            producer.start();
            for (int i = 0; i < 100; i++) {
                Message msg = new Message("TopicA", "TagA", ("Hello mq" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.println("send " + i + " , result:" + sendResult.getMsgId());
            }
            producer.shutdown();
        }
    }

## 消费者 ##

public class SimpleConsumer {
        public static void pushConsume(final String instanceName, final String group, final String topic, final String tag) throws Exception {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
            consumer.setNamesrvAddr("localhost:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe(topic, tag);
            consumer.setInstanceName(instanceName);
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.println("[" + instanceName + "," + group + "," + topic + "," + tag + "] consume: " + new String(msg.getBody()));
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
        }
    }

## 测试 ##

public static void main(String[] args) throws Exception {
    
            try {
                SimpleProducer.sendSync();
            } catch (Exception e) {
                e.printStackTrace();
            }
    
            Thread t2 = new Thread() {
                @Override
                public void run() {
                    try {
                        SimpleConsumer.pushConsume("A", "GroupA", "TopicA", "TagA");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            t2.start();
    
            Thread t3 = new Thread() {
                @Override
                public void run() {
                    try {
                        SimpleConsumer.pushConsume("B", "GroupA", "TopicA", "TagB");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            t3.start();
    
    
            t2.join();
            t3.join();
        }

### 结果： ###

....
    send 97 , result:AC1100013B2118B4AAC2808264CC0061
    send 98 , result:AC1100013B2118B4AAC2808264CD0062
    send 99 , result:AC1100013B2118B4AAC2808264CE0063
    [A,GroupA,TopicA,TagA] consume: Hello mq1
    [A,GroupA,TopicA,TagA] consume: Hello mq2
    [A,GroupA,TopicA,TagA] consume: Hello mq5
    ....

### 查看队列分布情况： ###

#### 消费者A ####

./bin/mqadmin consumerStatus -n "localhost:9876" -g "GroupA" -i "172.17.0.1@A"
    #Consumer MQ Detail#
    #Topic                            #Broker Name                      #QID  #ProcessQueueInfo   
    %RETRY%GroupA                     mo-x                              0     ProcessQueueInfo [commitOffset=0, cachedMsgMinOffset=0, cachedMsgMaxOffset=0, cachedMsgCount=0, cachedMsgSizeInMiB=0, transactionMsgMinOffset=0, transactionMsgMaxOffset=0, transactionMsgCount=0, locked=true, tryUnlockTimes=0, lastLockTimestamp=20180625231130503, droped=false, lastPullTimestamp=20180625231132580, lastConsumeTimestamp=20180625231129554]
    TopicA                            mo-x                              0     ProcessQueueInfo [commitOffset=50, cachedMsgMinOffset=0, cachedMsgMaxOffset=0, cachedMsgCount=0, cachedMsgSizeInMiB=0, transactionMsgMinOffset=0, transactionMsgMaxOffset=0, transactionMsgCount=0, locked=true, tryUnlockTimes=0, lastLockTimestamp=20180625231130503, droped=true, lastPullTimestamp=20180625231132552, lastConsumeTimestamp=20180625231129547]
    TopicA                            mo-x                              1     ProcessQueueInfo [commitOffset=50, cachedMsgMinOffset=0, cachedMsgMaxOffset=0, cachedMsgCount=0, cachedMsgSizeInMiB=0, transactionMsgMinOffset=0, transactionMsgMaxOffset=0, transactionMsgCount=0, locked=true, tryUnlockTimes=0, lastLockTimestamp=20180625231130503, droped=true, lastPullTimestamp=20180625231132572, lastConsumeTimestamp=20180625231129549]

#### 消费者B ####

./bin/mqadmin consumerStatus -n "localhost:9876" -g "GroupA" -i "172.17.0.1@B"
    #Consumer MQ Detail#
    #Topic                            #Broker Name                      #QID  #ProcessQueueInfo   
    TopicA                            mo-x                              2     ProcessQueueInfo [commitOffset=25, cachedMsgMinOffset=0, cachedMsgMaxOffset=0, cachedMsgCount=0, cachedMsgSizeInMiB=0, transactionMsgMinOffset=0, transactionMsgMaxOffset=0, transactionMsgCount=0, locked=true, tryUnlockTimes=0, lastLockTimestamp=20180625231150500, droped=false, lastPullTimestamp=20180625231209016, lastConsumeTimestamp=20180625231149571]
    TopicA                            mo-x                              3     ProcessQueueInfo [commitOffset=25, cachedMsgMinOffset=0, cachedMsgMaxOffset=0, cachedMsgCount=0, cachedMsgSizeInMiB=0, transactionMsgMinOffset=0, transactionMsgMaxOffset=0, transactionMsgCount=0, locked=true, tryUnlockTimes=0, lastLockTimestamp=20180625231150500, droped=false, lastPullTimestamp=20180625231209016, lastConsumeTimestamp=20180625231149565]

mo@mo-x:~/rocket-mq$ ./bin/mqadmin consumerProgress -n localhost:9876 -g "GroupA"
    Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0
    Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0
    #Topic                            #Broker Name                      #QID  #Broker Offset        #Consumer Offset      #Client IP           #Diff                 #LastTime
    %RETRY%GroupA                     mo-x                              0     0                     0                     172.17.0.1           0                     1970-01-01 08:00:00
    TopicA                            mo-x                              0     25                    25                    172.17.0.1           0                     2018-06-25 23:11:29
    TopicA                            mo-x                              1     25                    25                    172.17.0.1           0                     2018-06-25 23:11:29
    TopicA                            mo-x                              2     25                    25                    172.17.0.1           0                     2018-06-25 23:11:29
    TopicA                            mo-x                              3     25                    25                    172.17.0.1           0                     2018-06-25 23:11:29

## 测试结果总结： ##

*  1.生产者发送了100条TagA消息到TopicA
 *  2.消费者A和消费者B都在GroupA中，都订阅TopicA
 *  3.消费者A订阅TagA，消费者B订阅TagB
 *  4.消费者A收到了部分消息
 *  5.消费者A分配到了两个GroupA-TopicA的队列
 *  6.消费者B分配到了两个GroupA-TopicA的队列

## 总结： ##

[image-20210910230933439]: /images/20220828/9884ec50b31b456a8ceddd09c7b9db51.png
[image-20210910230959011]: /images/20220828/7ca435e03002459fa1d02d270fb48602.png