概述
rocketmq原理想必大家都有了解了，网上也有很多博客和资料讲述的很详细。本章主要是想讲一讲rocketmq中topic、group、tag之间的关系。
我一般喜欢带着问题去查看源码从而验证问题的结论。所以先来看看我的问题：
1、在consumer订阅消息中中允许topic、tag相同、group不同的消费者同时消费消息吗？
2、在consumer订阅消息中允许group、tag相同、topic不同的消费者同时消费消息吗？
3、在consumer订阅消息中允许group、topic相同、tag不同的消费者同时消费消息吗？
想要知道上面的问题。看看rocketmq中是如何实现订阅关系的吧。
1、订阅关系核心管理类方法：ConsumerManager#registerConsumer
public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {
//1、获取consumer组信息
ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
if (null == consumerGroupInfo) {
ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere);
ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp);
consumerGroupInfo = prev != null ? prev : tmp;
}
//2、更新消费客户端ip信息
boolean r1 =
consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
consumeFromWhere);
//3、更新消费端订阅的topic、tag等信息
boolean r2 = consumerGroupInfo.updateSubscription(subList);
if (r1 || r2) {
if (isNotifyConsumerIdsChangedEnable) {
this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
}
}
this.consumerIdsChangeListener.handle(ConsumerGroupEvent.REGISTER, group, subList);
return r1 || r2;
}
说明：consumerTable是一个Map类型变量，存放的是所有消费组信息，key存放的是groupName，value存放的是ConsumerGroupInfo组信息，我们再继续往下看consumerGroupInfo.updateSubscription方法，该方法是更新消费组信息的核心方法。
public boolean updateSubscription(final Set<SubscriptionData> subList) {
boolean updated = false;
//循环consumer订阅信息
for (SubscriptionData sub : subList) {
//根据topic获取订阅信息
SubscriptionData old = this.subscriptionTable.get(sub.getTopic());
//如果订阅信息不存在，则直接新增
if (old == null) {
SubscriptionData prev = this.subscriptionTable.putIfAbsent(sub.getTopic(), sub);
if (null == prev) {
updated = true;
log.info("subscription changed, add new topic, group: {} {}",
this.groupName,
sub.toString());
}
}
//如果订阅信息存在，判断新的订阅信息版本高于老的订阅信息(一般都是高于)
else if (sub.getSubVersion() > old.getSubVersion()) {
if (this.consumeType == ConsumeType.CONSUME_PASSIVELY) {
log.info("subscription changed, group: {} OLD: {} NEW: {}",
this.groupName,
old.toString(),
sub.toString()
);
}
//新的订阅信息覆盖老的订阅信息，这里可以看出来，同一个group和topic的情况下，tag不同，
//也会被覆盖掉，所以问题3的答案有了
this.subscriptionTable.put(sub.getTopic(), sub);
}
}
Iterator<Entry<String, SubscriptionData>> it = this.subscriptionTable.entrySet().iterator();
//这里循环判断subscriptionTable 与本次注册进来的subList比较
while (it.hasNext()) {
Entry<String, SubscriptionData> next = it.next();
String oldTopic = next.getKey();
boolean exist = false;
for (SubscriptionData sub : subList) {
//判断当前subscriptionTable中所有topic订阅信息是否都在新注册的列表中
if (sub.getTopic().equals(oldTopic)) {
exist = true;
break;
}
}
if (!exist) {
log.warn("subscription changed, group: {} remove topic {} {}",
this.groupName,
oldTopic,
next.getValue().toString()
);
//如果不存在，则删除group对应的topic订阅信息，
it.remove();
updated = true;
}
}
this.lastUpdateTimestamp = System.currentTimeMillis();
return updated;
}
现在我们来解答前面提的三个问题：
1、在consumer订阅消息中中允许topic、tag相同、group不同的消费者同时消费消息吗？
答：可以。因为在ConsumerManage#consumerTable 中是以groupName为key的，每个groupName对应的ConsumerGroupInfo相互隔离的。
2、在consumer订阅消息中允许group、tag相同、topic不同的消费者同时消费消息吗？
答：不可以。如下场景：
现有消费客户端consumer1，consumer2，topic1，topic2，group：group1，tag：tag1
consumer1订阅topic1，group1，tag1的订阅信息
consumer2订阅topic2，group1，tag1的订阅信息
步骤1:consumer1注册consumerGroupInfo信息调用updateSubscription方法更新subscriptionTableMap信息
步骤2:consumer2注册consumerGroupInfo信息调用updateSubscription方法时，如上述源码所示，因为consumer2只订阅了topic2，所以consumer1订阅的topic1订阅信息会被删除掉。
3、在consumer订阅消息中允许group、topic相同、tag不同的消费者同时消费消息吗？
答：不可以，如下场景：
现有消费客户端consumer1，consumer2，topic1，group：group1，tag：tag1，tag2
步骤1:consumer1订阅信息为：topic1，group1，tag1
步骤2:consumer2订阅信息为：topic1，group1，tag2，此时会更新订阅信息的时候会拿consumer2的订阅信息覆盖掉consumer1的订阅信息，具体代码请参考ConsumerGroupInfo#updateSubscription方法
那么。。。product中的group是用来干嘛的？ConsumerManage中的consumerGroupInfo信息从哪来的？
请听下回分解。rocketmq源码系列(4)-consumer启动过程的那些事
网友评论