一文讲透 RocketMQ 消费者是如何负载均衡的
Posted 勇哥java实战分享
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一文讲透 RocketMQ 消费者是如何负载均衡的相关的知识，希望对你有一定的参考价值。
RocketMQ 支持两种消息模式：集群消费（ Clustering ）和广播消费（ Broadcasting ）。
集群消费：同一 Topic 下的一条消息只会被同一消费组中的一个消费者消费。也就是说，消息被负载均衡到了同一个消费组的多个消费者实例上。
广播消费：当使用广播消费模式时，每条消息推送给集群内所有的消费者，保证消息至少被每个消费者消费一次。
我们重点讲解下集群消费的消费流程 ，因为集群消费是使用最普遍的消费模式，理解了集群消费，广播消费也就能顺理成章的掌握了。
集群消费示例代码里，启动消费者，我们需要配置三个核心属性：消费组名、订阅主题、消息监听器，最后调用 start 方法启动。
消费者启动后，我们可以将整个流程简化成：
4 负载均衡
消费端的负载均衡是指将 Broker 端中多个队列按照某种算法分配给同一个消费组中的不同消费者。
负载均衡是每个客户端独立进行计算，那么何时触发呢 ？
-
消费端启动时，立即进行负载均衡；
-
消费端定时任务每隔 20 秒触发负载均衡；
-
消费者上下线，Broker 端通知消费者触发负载均衡。
负载均衡流程如下：
1、发送心跳
消费者启动后，它就会通过定时任务不断地向 RocketMQ 集群中的所有 Broker 实例发送心跳包（消息消费分组名称、订阅关系集合、消息通信模式和客户端实例编号等信息）。
Broker 端在收到消费者的心跳消息后，会将它维护在 ConsumerManager 的本地缓存变量 consumerTable，同时并将封装后的客户端网络通道信息保存在本地缓存变量 channelInfoTable 中，为之后做消费端的负载均衡提供可以依据的元数据信息。
2、启动负载均衡服务
下图展示了按照主题负载均衡的代码片段：
负载均衡服务会根据消费模式为”广播模式”还是“集群模式”做不同的逻辑处理，这里主要来看下集群模式下的主要处理流程：
(1) 获取该主题下的消息消费队列集合；
(2) 查询 Broker 端获取该消费组下消费者 Id 列表；
(3) 先对 Topic 下的消息消费队列、消费者 Id 排序，然后用消息队列分配策略算法（默认为：消息队列的平均分配算法），计算出待拉取的消息队列；
这里的平均分配算法，类似于分页的算法，将所有 MessageQueue 排好序类似于记录，将所有消费端排好序类似页数，并求出每一页需要包含的平均 size 和每个页面记录的范围 range ，最后遍历整个 range 而计算出当前消费端应该分配到的记录。
(4) 分配到的消息队列集合与 processQueueTable 做一个过滤比对操作
消费者实例内 ，processQueueTable 对象存储着当前负载均衡的队列 ，以及该队列的消费快照。
标红的部分表示与分配到的消息队列集合互不包含，则需要将这些红色队列 Dropped 属性为 true , 然后从 processQueueTable 对象中移除。
绿色的部分表示与分配到的消息队列集合的交集，processQueueTable 对象中已经存在该队列。
黄色的部分表示这些队列需要添加到 processQueueTable 对象中，创建这些队列的消费快照。最后创建拉取消息请求列表，并将请求分发到消息拉取服务，进入拉取消息环节。
RocketMQ源码(16)—消费者负载均衡服务RebalanceService入口源码
基于RocketMQ release-4.9.3，深入的介绍了消费者负载均衡服务RebalanceService入口源码。
上一篇文章我们学习了RocketMQ源码(15)—消费者DefaultMQPushConsumer启动主要流程源码。
RocketMQ一个消费者组中可以有多个消费者，在集群模式下他们共同消费topic下的所有消息，RocketMQ规定一个消息队列仅能被一个消费者消费，但一个消费者可以同时消费多个消息队列。这就涉及到如何将多个消息队列分配给等多个消费者的问题。
RocketMQ中使用负载均衡服务RebalanceService来专门处理多个消息队列和消费者的对应关系，并且提供了多个不同的消费者负载均衡策略，即如何分配消息队列给这些消费者。
另外，当消费者正常退出，异常关闭通道，或者新加入的时候，同样需要负载均衡服务RebalanceService来进行消息队列分配的重平衡。
更重要的是，一个消费者启动之后，其消费消息的触发并不是pullMessageService消息拉取服务，而真正的源头正是负载均衡服务RebalanceService。因此我们有必要先学习RebalanceService的原理和源码。
文章目录
1 负载均衡or重平衡的触发
有三种情况会触发Consumer进行负载均衡或者说重平衡：
- RebalanceService服务是一个线程任务，由MQClientInstance启动，其每隔20s自动进行一次自动负载均衡。
- Broker触发的重平衡：
- Broker收到心跳请求之后如果发现消息中有新的consumer连接或者consumer订阅了新的topic或者移除了topic的订阅， 则Broker发送Code为NOTIFY_CONSUMER_IDS_CHANGED的请求给该group下面的所有Consumer，要求进行一次负载均衡。
- 如果某个客户端连接出现连接异常事件EXCEPTION、连接断开事件CLOSE、或者连接闲置事件IDLE，则Broker同样会发送重平衡请求给消费者组下面的所有消费者。
- 新的Consumer服务启动的时候，主动调用rebalanceImmediately唤醒负载均衡服务rebalanceService，进行重平衡。
1.1 RebalanceService自动重平衡
RebalanceService#run方法，也就是负载均衡服务运行的任务，最多每隔20s执行一次重平衡。主要逻辑是在mqClientFactory#doRebalance方法中实现的。
/**
* RebalanceServicede 方法
*/
@Override
public void run()
log.info(this.getServiceName() + " service started");
/*
* 运行时逻辑
* 如果服务没有停止，则在死循环中执行负载均衡
*/
while (!this.isStopped())
//等待运行，默认最多等待20s，可以被唤醒
this.waitForRunning(waitInterval);
//执行重平衡操作
this.mqClientFactory.doRebalance();
log.info(this.getServiceName() + " service end");
1.2 Consumer启动重平衡
新的Consumer服务启动的时候，主动调用rebalanceImmediately唤醒负载均衡服务rebalanceService，进行重平衡。
/**
* MQClientInstance的方法
* 立即重平衡
*/
public void rebalanceImmediately()
//唤醒重平衡服务，立即重平衡
this.rebalanceService.wakeup();
1.3 Broker请求重平衡
broker触发的重平衡有两种情况：
- Broker收到心跳请求之后如果发现消息中有新的consumer连接，或者consumer订阅了新的topic，或者移除了topic的订阅， 则Broker发送Code为NOTIFY_CONSUMER_IDS_CHANGED的请求给该group下面的所有Consumer，要求进行一次负载均衡。
- 如果某个客户端连接出现连接异常事件EXCEPTION、连接断开事件CLOSE、或者连接闲置事件IDLE，则Broker同样会发送重平衡请求给消费者组下面的所有消费者。处理入口方法为ClientHousekeepingService# doChannelCloseEvent方法。
新的Consumer和Producer启动的时候，就会发送心跳信息给Broker，MQClientInstance内部的服务也会定时30s发送心跳信息给Broker。关于发送心跳请求sendHeartbeatToAllBrokerWithLock方法的源码，我们在Producer启动的部分就讲过了，我们现在来看看Broker处理心跳请求的源码。
心跳请求的Code为HEART_BEAT，该请求最终被Broker的ClientManageProcessor处理器处理。
/**
* ClientManageProcessor的方法
*/
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
throws RemotingCommandException
switch (request.getCode())
//客户端心跳请求
case RequestCode.HEART_BEAT:
//客户端心跳请求
return this.heartBeat(ctx, request);
case RequestCode.UNREGISTER_CLIENT:
return this.unregisterClient(ctx, request);
case RequestCode.CHECK_CLIENT_CONFIG:
return this.checkClientConfig(ctx, request);
default:
break;
return null;
1.3.1 Broker处理心跳请求
Broker的ClientManageProcessor#heartBeat该方法用于Broker处理来自客户端（包括consumer和producer）的心跳请求。主要流程就是：
- 解码消息中的信息成为HeartbeatData对象，该对象的结构我们在在Producer启动的部分就讲过了。
- 循环遍历处理consumerDataSet集合，对ConsumerData信息进行注册或者更改，如果consumer信息发生了改变，Broker会发送NOTIFY_CONSUMER_IDS_CHANGED请求给同组的所有consumer客户端，要求进行重平衡操作。
- 循环遍历处理consumerDataSet集合，对 ProducerData信息进行注册或者更改。
/**
* ClientManageProcessor的方法
* <p>
* 处理客户端心跳请求
*/
public RemotingCommand heartBeat(ChannelHandlerContext ctx, RemotingCommand request)
//构建响应命令对象
RemotingCommand response = RemotingCommand.createResponseCommand(null);
//解码
HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
//构建客户端连接信息对象
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
ctx.channel(),
heartbeatData.getClientID(),
request.getLanguage(),
request.getVersion()
);
/*
* 1 循环遍历处理consumerDataSet，即处理consumer的心跳信息
*/
for (ConsumerData data : heartbeatData.getConsumerDataSet())
//查找broker缓存的当前消费者组的订阅组配置
SubscriptionGroupConfig subscriptionGroupConfig =
this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(
data.getGroupName());
boolean isNotifyConsumerIdsChangedEnable = true;
//如果已存在订阅组
if (null != subscriptionGroupConfig)
//当consumer发生改变的时候是否支持通知同组的所有consumer，默认true，即支持
isNotifyConsumerIdsChangedEnable = subscriptionGroupConfig.isNotifyConsumerIdsChangedEnable();
int topicSysFlag = 0;
if (data.isUnitMode())
topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
//尝试创建重试topic
String newTopic = MixAll.getRetryTopic(data.getGroupName());
this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
newTopic,
subscriptionGroupConfig.getRetryQueueNums(),
PermName.PERM_WRITE | PermName.PERM_READ, topicSysFlag);
/*
* 注册consumer，返回consumer信息是否已发生改变
* 如果发生了改变，Broker会发送NOTIFY_CONSUMER_IDS_CHANGED请求给同组的所有consumer客户端，要求进行重平衡操作
*/
boolean changed = this.brokerController.getConsumerManager().registerConsumer(
data.getGroupName(),
clientChannelInfo,
data.getConsumeType(),
data.getMessageModel(),
data.getConsumeFromWhere(),
data.getSubscriptionDataSet(),
isNotifyConsumerIdsChangedEnable
);
if (changed)
//如果consumer信息发生了改变，打印日志
log.info("registerConsumer info changed ",
data.toString(),
RemotingHelper.parseChannelRemoteAddr(ctx.channel())
);
/*
* 2 循环遍历处理producerDataSet，即处理producer的心跳信息
*/
for (ProducerData data : heartbeatData.getProducerDataSet())
/*
* 注册producer
*/
this.brokerController.getProducerManager().registerProducer(data.getGroupName(),
clientChannelInfo);
//返回响应
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
1.3.1.1 registerConsumer注册消费者
注册consumer，返回consumer信息是否已发生改变，如果发生了改变，Broker会发送NOTIFY_CONSUMER_IDS_CHANGED请求给同组的所有consumer客户端，要求进行重平衡操作。
/**
* ConsumerManager的方法
* <p>
* 注册consumer，返回consumer信息是否已发生改变
* 如果发生了改变，Broker会发送NOTIFY_CONSUMER_IDS_CHANGED请求给同组的所有consumer客户端，要求进行重平衡操作
*
* @param group 消费者组
* @param clientChannelInfo 客户端连接信息
* @param consumeType 消费类型，PULL or PUSH
* @param messageModel 消息模式，集群 or 广播
* @param consumeFromWhere 启动消费位置
* @param subList 订阅信息数据
* @param isNotifyConsumerIdsChangedEnable 一个consumer改变时是否通知该consumergroup中的所有consumer进行重平衡
* @return 是否重平衡
*/
public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable)
//获取当前group对应的ConsumerGroupInfo
ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
//如果为null，那么新建一个ConsumerGroupInfo并存入consumerTable
if (null == consumerGroupInfo)
ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere);
ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp);
consumerGroupInfo = prev != null ? prev : tmp;
/*
* 1 更新连接
*/
boolean r1 =
consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
consumeFromWhere);
/*
* 2 更新订阅信息
*/
boolean r2 = consumerGroupInfo.updateSubscription(subList);
/*
* 2 如果连接或者订阅信息有更新，并且允许通知，那么通知该consumergroup中的所有consumer进行重平衡
*/
if (r1 || r2)
if (isNotifyConsumerIdsChangedEnable)
//CHANGE事件
this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
//注册订阅信息到ConsumerFilterManager
this.consumerIdsChangeListener.handle(ConsumerGroupEvent.REGISTER, group, subList);
return r1 || r2;
1.3.1.1.1 updateChannel更新连接
更新此ConsumerGroup组对应的ConsumerGroupInfo的一些属性，并且还会判断当前连接是否是新连接，如果Broker此前没有该连接的信息，那么表示有新的consumer连接到此broker，那么需要通知当前ConsumerGroup的所有consumer进行重平衡。
/**
* ConsumerGroupInfo的方法
* <p>
* 更新连接
*
* @param infoNew 新连接信息
* @param consumeType 消费类型，PULL or PUSH
* @param messageModel 消息模式，集群 or 广播
* @param consumeFromWhere 启动消费位置
* @return 是否通知
*/
public boolean updateChannel(final ClientChannelInfo infoNew, ConsumeType consumeType,
MessageModel messageModel, ConsumeFromWhere consumeFromWhere)
boolean updated = false;
//更新信息
this.consumeType = consumeType;
this.messageModel = messageModel;
this.consumeFromWhere = consumeFromWhere;
//根据当前连接获取channelInfoTable缓存中的连接信息
ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getChannel());
//如果缓存中的连接信息为null，说明当前连接是一个新连接
if (null == infoOld)
//存入缓存
ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getChannel(), infoNew);
//长期按没有该连接信息，那么表示有新的consumer连接到此broekr，那么需要通知
if (null == prev)
log.info("new consumer connected, group: channel: ", this.groupName, consumeType,
messageModel, infoNew.toString());
updated = true;
infoOld = infoNew;
else
//异常情况
if (!infoOld.getClientId().equals(infoNew.getClientId()))
log.error("[BUG] consumer channel exist in broker, but clientId not equal. GROUP: OLD: NEW: ",
this.groupName,
infoOld.toString(),
infoNew.toString());
this.channelInfoTable.put(infoNew.getChannel(), infoNew);
//更新更新时间
this.lastUpdateTimestamp = System.currentTimeMillis();
infoOld.setLastUpdateTimestamp(this.lastUpdateTimestamp);
return updated;
1.3.1.1.2 updateSubscription更新订阅信息
更新此ConsumerGroup组对应的订阅信息集合，如果存在新增订阅的topic，或者移除了对于某个topic的订阅，那么需要通知当前ConsumerGroup的所有consumer进行重平衡。
该方法的大概步骤为：
- 该方法首先遍历当前请求传递的订阅信息集合，然后对于每个订阅的topic从subscriptionTable缓存中尝试获取，如果获取不到则表示新增了topic订阅信息，那么将新增的信息存入subscriptionTable。
- 然后遍历subscriptionTable集合，判断每一个topic是否存在于当前请求传递的订阅信息集合中，如果不存在，表示consumer移除了对于该topic的订阅，那么当前topic的订阅信息会从subscriptionTable集合中被移除。
这里的源码实际上很重要，他向我们传达出了什么信息呢？那就是RocketMQ需要保证组内的所有消费者订阅的topic都必须一致，否则就会出现订阅的topic被覆盖的情况。
根据刚才的源码分析，假设一个消费者组groupX里面有两个消费者，A消费者先启动并且订阅topicA，A消费者向broker发送心跳，那么subscriptionTable中消费者组groupX里面仅有topicA的订阅信息。
随后B消费者启动并且订阅topicB，B消费者也向broker发送心跳，那么根据该方法的源码，subscriptionTable中消费者组groupX里面的topicA的订阅信息将会被移除，而topicB的订阅信息会被存入进来。
这样就导致了topic订阅信息的相互覆盖，导致其中一个消费者能够消费消息，而另一个消费者不会消费。
/**
* ConsumerGroupInfo的方法
* 更新订阅信息
*
* @param subList 订阅信息集合
*/
public boolean updateSubscription(final Set<SubscriptionData> subList)
boolean updated = false;
//遍历订阅信息集合
for (SubscriptionData sub : subList)
//根据订阅的topic在ConsumerGroup的subscriptionTable缓存中此前的订阅信息
SubscriptionData old = this.subscriptionTable.get(sub.getTopic());
//如果此前没有关于该topic的订阅信息，那么表示此topic为新增订阅
if (old == null)
//存入subscriptionTable
SubscriptionData prev = this.subscriptionTable.putIfAbsent(sub.getTopic(), sub);
//此前没有关于该topic的订阅信息，那么表示此topic为新增订阅，那么需要通知
if (null == prev)
updated = true;
log.info("subscription changed, add new topic, group: ",
this.groupName,
sub.toString());
else if (sub.getSubVersion() > old.getSubVersion())
//更新数据
if (this.consumeType == ConsumeType.CONSUME_PASSIVELY)
log.info("subscription changed, group: OLD: NEW: ",
this.groupName,
old.toString(),
sub.toString()
);
this.subscriptionTable.put(sub.getTopic(), sub);
/*
* 遍历ConsumerGroup的subscriptionTable缓存
*/
Iterator<Entry<String, SubscriptionData>> it = this.subscriptionTable.entrySet().iterator();
while (it.hasNext())
Entry<String, SubscriptionData> next = it.next();
//获取此前订阅的topic
String oldTopic = next.getKey();
boolean exist = false;
//判断当前的subList是否存在该topic的订阅信息
for (SubscriptionData sub : subList)
//如果存在，则退出循环
if (sub.getTopic().equals(oldTopic))
exist = true;
break;
//当前的subList不存在该topic的订阅信息，说明consumer移除了对于该topic的订阅
if (!exist)
log.warn("subscription changed, group: remove topic ",
this.groupName,
oldTopic,
next.getValue().toString()
);
//移除数据
it.remove();
//那么需要通知
updated = true;
this.lastUpdateTimestamp = System.currentTimeMillis();
return updated;
1.3.1.1.3 consumerIdsChangeListener.handle监听器通知
该方法通知监听器处理对应的事件，需要进行通知的事件为ConsumerGroupEvent.CHANGE。
可以看到该方法中对于ConsumerGroupEvent.CHANGE事件的处理为：如果允许通知，则遍历该ConsumerGroup的连接集合，然后对每个连接调用notifyConsumerIdsChanged方法通知对应的客户端消费者执行负载均衡。
/**
* DefaultConsumerIdsChangeListener的方法
*
* 处理监听到的事件
* @param event 事件
* @param group 消费者组
* @param args 参数
*/
@Override
public void handle(ConsumerGroupEvent event, String group, Object... args)
if (event == null)
return;
switch (event)
//改变事件，需要通知该消费者组的每一个消费者
case CHANGE:
if (args == null || args.length < 1)
return;
//获取参数
List<Channel> channels = (List<Channel>) args[0];
//如果允许通知
if (channels != null && brokerController.getBrokerConfig().isNotifyConsumerIdsChangedEnable())
//遍历连接集合
for (Channel chl : channels)
//通知该消费者客户端执行负载均衡
this.brokerController.getBroker2Client()一文讲透Dubbo负载均衡之最小活跃数算法
RocketMQ源码(17)—RebalanceService消费者负载均衡过程源码