前言
本文介绍了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对象可以订阅多个topic Mapsub = this.defaultMQPushConsumer.getSubscription(); if (sub != null) { for (final Map.Entry 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线程体:
@Override public 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:
@Override public void doRebalance() { if (this.rebalanceImpl != null) { this.rebalanceImpl.doRebalance(); } }
步骤II 中完善了rebalanceImpl实例,为调用rebalanceImpl.doRebalance()提供了初始数据。
rebalanceImpl.doRebalance()过程如下:
public void doRebalance() { // 前文copySubscription中初始化了SubscriptionInner MapsubTable = this.getSubscriptionInner(); if (subTable != null) { for (final Map.Entry 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相关的所有消息队列 SetmqSet = this.topicSubscribeInfoTable.get(topic); //2. 从broker端获取消费该消费组的所有客户端clientId List cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup); f (null == mqSet) { ... } if (null == cidAll) { ... } if (mqSet != null && cidAll != null) { List mqAll = new ArrayList (); mqAll.addAll(mqSet); Collections.sort(mqAll); Collections.sort(cidAll); // 3.创建DefaultMQPushConsumer对象时默认设置为AllocateMessageQueueAveragely AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy; List 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 allocateResultSet = new HashSet (); if (allocateResult != null) { allocateResultSet.addAll(allocateResult); } 、 //6. 更新updateProcessQueue boolean 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下所有在线客户端列表中。
当前消费对进行负载均衡策略后获取对应的消息消费队列。具体的算法很简单,可以看源码。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。