rocketmq源码解析 rocketmq源码部署( 四 ) _云知道

1.2 RebalanceService#run：启动消息端负载均衡服务。
// RebalanceService#run public void run() { log.info(this.getServiceName() + " service started"); while (!this.isStopped()) { this.waitForRunning(waitInterval); this.mqClientFactory.doRebalance(); } log.info(this.getServiceName() + " service end"); } // MQClientInstance#doRebalance public void doRebalance() { for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) { MQConsumerInner impl = entry.getValue(); if (impl != null) { try { impl.doRebalance(); } catch (Throwable e) { log.error("doRebalance exception", e); } } } } // DefaultMQPushConsumerImpl#doRebalance public void doRebalance() { if (!this.pause) { this.rebalanceImpl.doRebalance(this.isConsumeOrderly()); } }1.2.1.1.1 RebalanceImpl#doRebalance：负载均衡服务类处理。
// RebalanceImpl#doRebalance public void doRebalance(final boolean isOrder) { Map<String, SubscriptionData> subTable = this.getSubscriptionInner(); if (subTable != null) { for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) { final String topic = entry.getKey(); try { this.rebalanceByTopic(topic, isOrder); } catch (Throwable e) { if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { log.warn("rebalanceByTopic Exception", e); } } } } this.truncateMessageQueueNotMyTopic(); } // RebalanceImpl#rebalanceByTopic switch (messageModel) { case BROADCASTING: { Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic); if (mqSet != null) { boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);// 根据Toipc去除queue if (changed) { this.messageQueueChanged(topic, mqSet, mqSet); log.info("messageQueueChanged {} {} {} {}", consumerGroup, topic, mqSet, mqSet); } } else { ...... // RebalanceImpl#updateProcessQueueTableInRebalance this.dispatchPullRequest(pullRequestList);// RebalancePushImpl分发消息1.2.1.1.1.1.1.1 RebalancePushImpl#dispatchPullRequest：RebalancePushImpl分发。
// RebalancePushImpl#dispatchPullRequest public void dispatchPullRequest(List<PullRequest> pullRequestList) { for (PullRequest pullRequest : pullRequestList) { this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest); log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest); } }五.总结
相比Producer的发送流程，Consumer的接收流程稍微复杂一点。通过上面的源码分析，可以知道RocketMQ是怎样保证消息的有序的：
1.通过ReblanceImp的lockAll方法，每隔一段时间定时锁住当前消费端正在消费的队列。设置本地队列ProcessQueue的locked属性为true 。保证broker中的每个消息队列只对应一个消费端；
2.另外，消费端也是通过锁，保证每个ProcessQueue只有一个线程消费。

rocketmq源码解析 rocketmq源码部署( 四 )

推荐阅读

如何改掉忧虑的习惯？

QQ空间相册之间怎么合并啊

山鸡和家鸡能繁殖后代吗?

越剧名段有哪些

香菇炒肉片的家常做法

最便宜的狗狗品种排名最便宜的狗

暖橘色配什么颜色好看

中文 Cisco4006交换机简明配置手册

2022重庆公租房网上申请网址 2022重庆公租房网上申请网址是什么

qq空间怎么设置半年可见

大乐斗什么时候更新,超级大乐透品牌形象由你定

发热严重！！！！！

iphone屏幕绿屏了怎么办，绿屏iphone12是什么屏幕

天地诸神轮回爆什么,本神子要让你体会

瑜伽tv效果好怎么样,每日瑜伽app怎么样

金蛋理财app能进了吗