本文介紹了DefaultMQPushConsumerImpl消費者，客戶端負載均衡相關知識點。本文從DefaultMQPushConsumerImpl啟動過程到實現負載均衡，從源代碼一步一步分析，共分為6個部分進行介紹，其中第6個部分?rebalanceByTopic 為負載均衡的核心邏輯模塊，具體過程運用了圖文進行闡述。 介紹之前首先拋出幾個問題： 1. 要做負載均衡，首先要解決的一個問題是什么？ 2. 負載均衡是Client端處理還是Broker端處理？ 個人理解： 1. 要做負載均衡，首先要做的就是信號收集。 所謂信號收集，就是得知道每一個consumerGroup有哪些consumer，對應的topic是誰。信號收集分為Client端信號收集與Broker端信號收集兩個部分。 2. 負載均衡放在Client端處理。 具體做法是：消費者客戶端在啟動時完善rebalanceImpl實例，同時拷貝訂閱信息存放rebalanceImpl實例對象中，另外也是很重要的一個步驟 -- 通過心跳消息，不停的上報自己到所有Broker，注冊RegisterConsumer，等待上述過程準備好之后在Client端不斷執行的負載均衡服務線程從Broker端獲取一份全局信息（該consumerGroup下所有的消費Client），然后分配這些全局信息，獲取當前客戶端分配到的消費隊列。 本文具體的內容： I. copySubscription Client端信號收集，拷貝訂閱信息。 在DefaultMQPushConsumerImpl.start()時，會將消費者的topic訂閱關系設置到rebalanceImpl的SubscriptionInner的map中用于負載： private void copySubscription() throws MQClientException {try {//注：一個consumer對象可以訂閱多個topicMap<String, String> sub = this.defaultMQPushConsumer.getSubscription();if (sub != null) {for (final Map.Entry<String, String> entry : sub.entrySet()) {final String topic = entry.getKey();final String subString = entry.getValue();SubscriptionData subscriptionData =FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),// topic, subString);this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);}}if (null == this.messageListenerInner) {this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();}switch (this.defaultMQPushConsumer.getMessageModel()) {case BROADCASTING:break;case CLUSTERING:final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());SubscriptionData subscriptionData =FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),// retryTopic, SubscriptionData.SUB_ALL);this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);break;default:break;}}catch (Exception e) {throw new MQClientException("subscription exception", e);}} FilterAPI.buildSubscriptionData接口將訂閱關系轉換為SubscriptionData 數據，其中subString包含訂閱tag等信息。另外，如果該消費者的消費模式為集群消費，則會將retry的topic一并放到。 II. 完善rebalanceImpl實例 Client繼續收集信息： this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup()); this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());this.rebalanceImpl.setmQClientFactory(this.mQClientFactory); 本文以DefaultMQPushConsumerImpl為例，因此this對象類型為DefaultMQPushConsumerImp。 III. this.rebalanceService.start() 開啟負載均衡服務。this.rebalanceService是一個RebalanceService實例對象，它繼承與ServiceThread，是一個線程類。 this.rebalanceService.start()執行時，也即執行RebalanceService線程體： @Overridepublic void run() {log.info(this.getServiceName() + " service started");while (!this.isStoped()) {this.waitForRunning(WaitInterval);this.mqClientFactory.doRebalance();}log.info(this.getServiceName() + " service end");}

?

IV. this.mqClientFactory.doRebalance 客戶端遍歷消費組table，對該客戶端上所有消費者獨立進行負載均衡，分發消費隊列： public void doRebalance() {for (String group : this.consumerTable.keySet()) {MQConsumerInner impl = this.consumerTable.get(group);if (impl != null) {try {impl.doRebalance();} catch (Exception e) {log.error("doRebalance exception", e);}}}}

?

V. MQConsumerInner.doRebalance 由于本文以DefaultMQPushConsumerImpl消費過程為例，即DefaultMQPushConsumerImpl.doRebalance： @Overridepublic void doRebalance() {if (this.rebalanceImpl != null) {this.rebalanceImpl.doRebalance();}} 步驟II 中完善了rebalanceImpl實例，為調用rebalanceImpl.doRebalance()提供了初始數據。 rebalanceImpl.doRebalance()過程如下： public void doRebalance() {// 前文copySubscription中初始化了SubscriptionInnerMap<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);} catch (Exception e) {if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {log.warn("rebalanceByTopic Exception", e);}}}}this.truncateMessageQueueNotMyTopic();}

?

VI. rebalanceByTopic -- 核心步驟之一
rebalanceByTopic方法中根據消費者的消費類型為BROADCASTING或CLUSTERING做不同的邏輯處理。CLUSTERING邏輯包括BROADCASTING邏輯，本部分只介紹集群消費負載均衡的邏輯。
集群消費負載均衡邏輯主要代碼如下（省略了log等代碼）：

//1.從topicSubscribeInfoTable列表中獲取與該topic相關的所有消息隊列 Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic); //2. 從broker端獲取消費該消費組的所有客戶端clientId List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);f (null == mqSet) { ... } if (null == cidAll) { ... } if (mqSet != null && cidAll != null) {List<MessageQueue> mqAll = new ArrayList<MessageQueue>();mqAll.addAll(mqSet);Collections.sort(mqAll);Collections.sort(cidAll);// 3.創建DefaultMQPushConsumer對象時默認設置為AllocateMessageQueueAveragelyAllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;List<MessageQueue> allocateResult = null;try {// 4.調用AllocateMessageQueueAveragely.allocate方法，獲取當前client分配消費隊列allocateResult = strategy.allocate(this.consumerGroup, this.mQClientFactory.getClientId(), mqAll,cidAll);} catch (Throwable e) {return;}// 5. 將分配得到的allocateResult 中的隊列放入allocateResultSet 集合Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();if (allocateResult != null) {allocateResultSet.addAll(allocateResult);} 、//6. 更新updateProcessQueueboolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet);if (changed) {this.messageQueueChanged(topic, mqSet, allocateResultSet);} } 注：BROADCASTING邏輯只包含上述的1、6。 集群消費負載均衡邏輯中的1、2、4這三個點相關知識為其核心過程，各個點相關知識如下： 第1點：從topicSubscribeInfoTable列表中獲取與該topic相關的所有消息隊列

?

第2點： 從broker端獲取消費該消費組的所有客戶端clientId 首先，消費者對象不斷地向所有broker發送心跳包，上報自己，注冊并更新訂閱關系以及客戶端ChannelInfoTable；之后，客戶端在做消費負載均衡時獲取那些消費客戶端，對這些客戶端進行負載均衡，分發消費的隊列。具體過程如下圖所示：

第4點：調用AllocateMessageQueueAveragely.allocate方法，獲取當前client分配消費隊列

?

注：上圖中cId1、cId2、...、cIdN通過?getConsumerIdListByGroup 獲取，它們在這個ConsumerGroup下所有在線客戶端列表中。 當前消費對進行負載均衡策略后獲取對應的消息消費隊列。具體的算法很簡單，可以看源碼。

?

轉載于:https://www.cnblogs.com/chenjunjie12321/p/7913323.html

總結

以上是生活随笔為你收集整理的rocketmq消费负载均衡--push消费为例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。