这篇文章主要介绍RocketMQ中push consumer启动之触发消息拉取的示例代码，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！
成都创新互联主营彝良网站建设的网络公司,主营网站建设方案,成都App定制开发,彝良h5重庆小程序开发公司搭建,彝良网站营销推广欢迎彝良等地区企业咨询
push consumer启动方法DefaultMQPushConsumerImpl.start最后一步会触发MQClientInstance.rebalanceImmediately，该调用最终会进入到RebalanceImpl.doRebalance中，它会根据topic当前的实际consumer数量（从nameserver获取）通过负载均衡原则来决定自己所要订阅的message queue。然后在本地创建对应的消息缓存队列（ProcessQueue），并触发消息拉取操作。
RebalanceImpl是整个consumer的核心，它即保存本消费者订阅的topic信息，又缓存了topic中的message数据。RebalanceImpl相关的几个核心类如下：
MessageQueue代表的是远端broker上一个topic下的某个message queue
ProcessQueue是对远端message queue的一个本地缓存，拉取下来的消息都存在一个TreeMap中，其中key是commitlog中的offset
RebalanceImpl中保存了三种关系：message queue和process queue的映射关系；topic和message queue的映射关系；topic的订阅关系
doRebalance方法会调用rebalanceByTopic来决定本消费者具体要订阅一个topic下的哪些message queue，以达到负载均衡的效果。
private void rebalanceByTopic(final String topic, final boolean isOrder) {
switch (messageModel) {
case BROADCASTING: {
// 广播模式，订阅所有message queue
Set mqSet = this.topicSubscribeInfoTable.get(topic);
if (mqSet != null) {
boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
if (changed) {
this.messageQueueChanged(topic, mqSet, mqSet);
log.info("messageQueueChanged {} {} {} {}",
consumerGroup,
topic,
mqSet,
mqSet);
}
} else {
log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
}
break;
}
case CLUSTERING: {
// 集群模式，获取该topic下所有的message queue + 该topic所有的consumer
Set mqSet = this.topicSubscribeInfoTable.get(topic);
List cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
if (null == mqSet) {
if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
}
}
if (null == cidAll) {
log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
}
if (mqSet != null && cidAll != null) {
List mqAll = new ArrayList();
mqAll.addAll(mqSet);
Collections.sort(mqAll);
Collections.sort(cidAll);
AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
List allocateResult = null;
try {
// 通过负载均衡策略计算出当前消费者所需订阅的message queue子集
allocateResult = strategy.allocate(
this.consumerGroup,
this.mQClientFactory.getClientId(),
mqAll,
cidAll);
} catch (Throwable e) {
log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
e);
return;
}
Set allocateResultSet = new HashSet();
if (allocateResult != null) {
allocateResultSet.addAll(allocateResult);
}
// 更新Process Queue缓存列表
boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
if (changed) {
log.info(
"rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, clientId={}, mqAllSize={}, cidAllSize={}, rebalanceResultSize={}, rebalanceResultSet={}",
strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), mqSet.size(), cidAll.size(),
allocateResultSet.size(), allocateResultSet);
this.messageQueueChanged(topic, mqSet, allocateResultSet);
}
}
break;
}
default:
break;
}
} rebalanceByTopic中通过负载均衡策略计算出当前消费者对于一个topic实际订阅的message queue子集之后，就会在updateProcessQueueTableInBalance方法中创建ProcessQueue，并启动消息拉取。
