其他分享
首页 > 其他分享> > RocketMQ入门到入土(七 )为什么同一个消费组设置不同tag会出现奇怪现象

RocketMQ入门到入土(七 )为什么同一个消费组设置不同tag会出现奇怪现象

作者:互联网

RocketMQ入门到入土(七 )为什么同一个消费组设置不同tag会出现奇怪现象

精彩推荐

一百期Java面试题汇总
SpringBoot内容聚合
IntelliJ IDEA内容聚合
Mybatis内容聚合

接上一篇:RocketMQ入门到入土(六)发消息的时候选择queue的算法有哪些?

一、问题复现
1、描述
两个一样的Consumer Group的Consumer订阅同一个Topic,但是是不同的tag,Consumer1订阅Topic的tag1,Consumer2订阅Topic的tag2,然后分别启动。这时候往Topic的tag1里发送10条数据,Topic的tag2里发送10条。目测应该是Consumer1和Consumer2分别收到对应的10条消息。结果却是只有Consumer2收到了消息,而且只收到了4-6条消息,不固定。

2、代码
2.1、Consumer

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer");
        consumer.setNamesrvAddr("124.57.180.156:9876");
        consumer.subscribe("TopicTest2","tag1");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                System.out.println(msg.getTags());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("ConsumerStarted.");
    }
}

启动这个订阅了TopicTest2的tag1标签的Consumer,然后将tag1改为tag2再次启动Consumer。这就相当于启动了两个Consumer进程,一个订阅了TopicTest2的tag1标签,另一个订阅了TopicTest2的tag2标签。

2.2、Producer


public class Producer {
    public static void main(String[] args) throws MQClientException {
        final DefaultMQProducer producer = new DefaultMQProducer("test-producer");
        producer.setNamesrvAddr("124.57.180.156:9876");
        producer.start();

        for (int i = 0; i < 10; i++){
            try {
                Message msg = new Message("TopicTest2", "tag1", ("Hello tag1 - "+i).getBytes());
                SendResult sendResult = producer.send(msg);
                System.out.println(sendResult);
            }catch(Exception e) {
                e.printStackTrace();
            }
        }
    }
}

启动Producer,往TopicTest2的tag1里发10条消息。再次将tag1改为tag2,然后再次启动Producer进行发送,这样就是TopicTest2的tag1下有10条消息,TopicTest2的tag2下也有10条消息。

3、结果
Consumer和Producer都启动后,发现如下:

那是因为:你是集群模式消费,它会负载均衡分配到各个节点去消费,所以一半消息(不固定个数)跑到了Consumer1上,结果Consumer1订阅的是tag1,所以不会任何输出。

如果换成BROADCASTING,那绝逼后者会收到全部消息,而不是一半,因为广播是广播全部Consumer。

三、源码验证
1、调用链

# 核心在于如下这个方法
org.apache.rocketmq.broker.client.ConsumerManager#registerConsumer()

# 关键调用链如下
# 入口是Broker启动的时候
org.apache.rocketmq.broker.BrokerStartup#start()
org.apache.rocketmq.broker.BrokerController#start()
org.apache.rocketmq.remoting.netty.NettyRemotingServer#start() 
org.apache.rocketmq.remoting.netty.NettyRemotingServer#prepareSharableHandlers()
org.apache.rocketmq.remoting.netty.NettyRemotingServer.NettyServerHandler#channelRead0()
org.apache.rocketmq.remoting.netty.NettyRemotingAbstract#processMessageReceived()
org.apache.rocketmq.remoting.netty.NettyRemotingAbstract#proce***equestCommand()
org.apache.rocketmq.broker.processor.ClientManageProcessor#proce***equest()
org.apache.rocketmq.broker.processor.ClientManageProcessor#heartBeat()
org.apache.rocketmq.broker.client.ConsumerManager#registerConsumer()

2、源码
2.1、registerConsumer


/**
 * Consumer信息
 */
public class ConsumerGroupInfo {
    // 组名
    private final String groupName;
    // topic信息,比如topic、tag等
    private final ConcurrentMap<String/* Topic */, SubscriptionData> subscriptionTable =
        new ConcurrentHashMap<String, SubscriptionData>();
    // 客户端信息,比如clientId等
    private final ConcurrentMap<Channel, ClientChannelInfo> channelInfoTable =
        new ConcurrentHashMap<Channel, ClientChannelInfo>(16);
    // PULL/PUSH
    private volatile ConsumeType consumeType;
    // 消费模式:BROADCASTING/CLUSTERING
    private volatile MessageModel messageModel;
    // 消费到哪了
    private volatile ConsumeFromWhere consumeFromWhere;
}

/**
 * 通过心跳将Consumer信息注册到Broker端。
 */
public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
        ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
        final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {

    // consumerTable:维护所有的Consumer
    ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
    // 如果没有Consumer,则put到map里
    if (null == consumerGroupInfo) {
        ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere);
        // put到map里
        ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp);
        consumerGroupInfo = prev != null ? prev : tmp;
    }

    // 更新Consumer信息,客户端信息
    boolean r1 =
        consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
                                        consumeFromWhere);
    // 更新订阅Topic信息
    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;
}

从这一步可以看出消费者信息是以groupName为key,ConsumerGroupInfo为value存到map(consumerTable)里的,那很明显了,后者肯定会覆盖前者的,因为key是一样的。而后者的tag是tag2,那肯定覆盖了前者的tag1,这部分是存到ConsumerGroupInfo的subscriptionTable里面的

private final ConcurrentMap<String/* Topic */, SubscriptionData> subscriptionTable =
    new ConcurrentHashMap<String, SubscriptionData>();
SubscriptionData包含了topic等信息

public class SubscriptionData implements Comparable<SubscriptionData> {
    // topic
    private String topic;
    private String subString;
    // tags
    private Set<String> tagsSet = new HashSet<String>();
    private Set<Integer> codeSet = new HashSet<Integer>();
}

2.2、两个问题
1.topic、tag等信息是怎么覆盖的?

boolean r1 = consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,consumeFromWhere);
/**
 * 其实很简单,就是以topic为key,SubscriptionData为value。而SubscriptionData里包含了tags信息,所以直接覆盖掉
 */
public boolean updateSubscription(final Set<SubscriptionData> subList) {
    for (SubscriptionData sub : subList) {
        SubscriptionData old = this.subscriptionTable.get(sub.getTopic());
        if (old == null) {
            SubscriptionData prev = this.subscriptionTable.putIfAbsent(sub.getTopic(), sub);
        } else if (sub.getSubVersion() > old.getSubVersion()) {
            this.subscriptionTable.put(sub.getTopic(), sub);
        }
    }
}

等等,这里好像有新发现ConsumerGroupInfo#subscriptionTable

// {@link org.apache.rocketmq.broker.client.ConsumerGroupInfo#subscriptionTable}
private final ConcurrentMap<String/* Topic */, SubscriptionData> subscriptionTable =
        new ConcurrentHashMap<String, SubscriptionData>();

可以有意外收获就是topic作为map的key,那岂不是一个Consumer可以订阅多个Topic?是的,通过这段源码可以发现是没毛病的,我也测试过。

2.这么看的话Consumer端只会存在一个进程,因为同组,注册进去就覆盖了呀?

大哥,注意ConsumerGroupInfo里的channelInfoTable


// 客户端信息,比如clientId等
private final ConcurrentMap<Channel, ClientChannelInfo> channelInfoTable =
    new ConcurrentHashMap<Channel, ClientChannelInfo>(16);

ClientChannelInfo是包含clientId等信息的,代表一个Consumer。注册方法是:

boolean r2 = consumerGroupInfo.updateSubscription(subList);
/**
 * 下面是删减后的代码,其实就是以Channel作为key,每个Consumer的Channel是不一样的。所以能存多个Consumer客户端
 */
public boolean updateChannel(final ClientChannelInfo infoNew, ConsumeType consumeType,
        MessageModel messageModel, ConsumeFromWhere consumeFromWhere) {
    ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getChannel());
    if (null == infoOld) {
        ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getChannel(), infoNew);
    }
}

标签:tag1,ConsumerGroupInfo,private,入土,tag,apache,new,Consumer,RocketMQ
来源: https://blog.51cto.com/u_15127600/2753088