RocketMQ源码(16)—消费者负载均衡服务RebalanceService入口源码
Posted 刘Java
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了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()RocketMQ源码(17)—RebalanceService消费者负载均衡过程源码