- RocketMQ一个消费者组中可以有多个消费者, 在集群模式下他们共同消费topic下的所有消息, RocketMQ规定一个消息队列仅能被一个消费者消费, 但是一个消费者可以同时消费多个消息队列。
- 需要负载均衡服务RebalanceService来进行消息队列分配的重平衡。
- 使用负载均衡服务RebalanceService来专门处理多个消息队列和消费者的对应关系, 如何分配消息队列给这些消费者。
RocketMQ消费者负载均衡服务RebalanceService的入口代码
1.负载均衡或者重平衡
有三种情况会触发Consumer进行负载均衡或者说重平衡。
1、 RebalanceService服务是一个线程任务,由MQClientInstance启动,每隔20s自动进行一次自动负载均衡；
2、 Broker触发的重平衡:；
1、 Broker收到心跳请求之后如果发现消息中有新的consumer连接或者consumer订阅了新的topic或者移除了topic的订阅,Broker发送Code为NOTIFY_CONSUMER_IDS_CHANGED的请求给该group下面的所有Consumer,要求进行一次负载均衡；
2、 如果某个客户端连接出现连接异常事件EXCEPTION、连接断开事件CLOSE、或者连接闲置事件IDLE,则Broker同样会发送重平衡请求给消费者组下面的所有消费者；
3、 新的Consumer服务启动的时候,主动调用rebalanceImmediately唤醒负载均衡服务rebalanceService；
1.1 RebalanceService自动重平衡
RebalanceService#run方法, 最多每隔20s执行一次重平衡。
/**
* 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请求重平衡
1、 Broker收到心跳请求之后如果发现消息中有新的consumer连接或者consumer订阅了新的topic或者移除了topic的订阅,Broker发送Code为NOTIFY_CONSUMER_IDS_CHANGED的请求给该group下面的所有Consumer,要求进行一次负载均衡；
2. 如果某个客户端连接出现连接异常事件EXCEPTION、连接断开事件CLOSE、或者连接闲置事件IDLE, 则Broker同样会发送重平衡请求给消费者组下面的所有消费者。处理入口方法为ClientHousekeepingService# doChannelCloseEvent方法。
新的Consumer和Producer启动的时候, 发送心跳给Broker, MQClientInstance的内部的服务也会定时30s发送心跳信息给Broker。
心跳请求的Code为HEART_BEAT, 该请求最终被Broker的ClientManageProcessor处理器处理。
/**
* ClientManageProcessor的方法, Broker处理心跳方法
*/
@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.4 Broker处理心跳请求
Broker的ClientManageProcessor#heartBeat该方法用于Broker处理来自客户端, consumer和producer的请求。
1、 解码消息中的信息成为HeartbeatData对象；
2、 循环遍历处理consumerDataSet集合,对ConsumerData信息进行注册或者更改,如果consumer信息修改的话,Broker会发送NOTIFY_CONSUMER_IDS_CHANGED请求给同组的所有consumer客户端,进行重平衡操作；
3、 循环遍历处理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.4.1 registerConsumer注册消费者#
循环遍历处理consumerDataSet集合, 对ConsumerData信息进行注册或者更改, 如果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());
}
}
// 3.注册订阅信息到ConsumerFilterManager
this.consumerIdsChangeListener.handle(ConsumerGroupEvent.REGISTER, group, subList);
return r1 || r2;
}
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;
}
2、 updateSubscription更新订阅信息；
更新此ConsumerGroup组对应的订阅信息集合, 如果存在新增订阅的topic, 或者移除某个topic的订阅, 那么需要通知当前ConsumerGroup的所有consumer进行重平衡。
RocketMQ需要保证组内的所有消费者订阅的topic都必须一致。
1、 方法首先遍历当前请求传递的订阅信息集合,然后对于每个订阅的topic从subscriptionTable缓存中尝试获取,获取不到则表示新增了topic订阅信息,存入subscriptionTable；
2、 遍历subscriptionTable集合,判断每一个topic是否存在于当前请求传递的订阅信息集合中,如果不存在,表示consumer移除了topic的订阅,那么当前topic的订阅信息会从subscriptionTable集合中被移除；
/**
* 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;
}
3、 consumerIdsChangeListener.handle监听器通知；
通知监听器处理对应的事件, 事件为ConsumerGroupEvent.CHANGE。
1、 如果运行通知,则遍历该ConsumerGroup的连接集合,然后对每个连接调用notifyConsumerIdsChanged方法通知对应的客户端消费者执行负载均衡；
2、 notifyConsumerIdsChanged:broker发送客户端一个重平衡请求,Code为NOTIFY_CONSUMER_IDS_CHANGED；
/**
* 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().notifyConsumerIdsChanged(chl, group);
}
}
break;
case UNREGISTER:
this.brokerController.getConsumerFilterManager().unRegister(group);
break;
case REGISTER:
if (args == null || args.length < 1) {
return;
}
Collection<SubscriptionData> subscriptionDataList = (Collection<SubscriptionData>) args[0];
this.brokerController.getConsumerFilterManager().register(group, subscriptionDataList);
break;
default:
throw new RuntimeException("Unknown event " + event);
}
}
/**
* Broker2Client的方法
* <p>
* 通知消费者变更
*
* @param channel 连接
* @param consumerGroup 消费者组
*/
public void notifyConsumerIdsChanged(
final Channel channel,
final String consumerGroup) {
if (null == consumerGroup) {
log.error("notifyConsumerIdsChanged consumerGroup is null");
return;
}
//构建请求头
NotifyConsumerIdsChangedRequestHeader requestHeader = new NotifyConsumerIdsChangedRequestHeader();
requestHeader.setConsumerGroup(consumerGroup);
//构建远程命令对象，请求code为NOTIFY_CONSUMER_IDS_CHANGED
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader);
try {
//发送单向请求，无需等待客户端回应
this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
} catch (Exception e) {
log.error("notifyConsumerIdsChanged exception. group={}, error={}", consumerGroup, e.toString());
}
}
1.4.2 客户端处理重平衡请求#
broker的请求在客户端是通过ClientRemotingProcessor#processRequest处理的。
NOTIFY_CONSUMER_IDS_CHANGED请求通过客户端的ClientRemotingProcessor#notifyConsumerIdsChanged方法处理。
/**
* ClientRemotingProcessor的方法
* <p>
* 处理来自远程服务端的请求
*/
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
switch (request.getCode()) {
case RequestCode.CHECK_TRANSACTION_STATE:
return this.checkTransactionState(ctx, request);
case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED:
//处理NOTIFY_CONSUMER_IDS_CHANGED请求
return this.notifyConsumerIdsChanged(ctx, request);
case RequestCode.RESET_CONSUMER_CLIENT_OFFSET:
return this.resetOffset(ctx, request);
case RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT:
return this.getConsumeStatus(ctx, request);
case RequestCode.GET_CONSUMER_RUNNING_INFO:
return this.getConsumerRunningInfo(ctx, request);
case RequestCode.CONSUME_MESSAGE_DIRECTLY:
return this.consumeMessageDirectly(ctx, request);
case RequestCode.PUSH_REPLY_MESSAGE_TO_CLIENT:
return this.receiveReplyMessage(ctx, request);
default:
break;
}
return null;
}
notifyConsumerIdsChanged客户端重平衡
客户端接口到broker的重平衡请求之后, 调用这个方法。内部仅仅是调用我们之前讲的rebalanceImmediately 方法唤醒负载均衡服务rebalanceService, 进行重平衡。
public RemotingCommand notifyConsumerIdsChanged(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
try {
//解析请求头
final NotifyConsumerIdsChangedRequestHeader requestHeader =
(NotifyConsumerIdsChangedRequestHeader) request.decodeCommandCustomHeader(NotifyConsumerIdsChangedRequestHeader.class);
//打印日志
log.info("receive broker's notification[{}], the consumer group: {} changed, rebalance immediately",
RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
requestHeader.getConsumerGroup());
//熟悉的方法，立即进行重平衡
this.mqClientFactory.rebalanceImmediately();
} catch (Exception e) {
log.error("notifyConsumerIdsChanged exception", RemotingHelper.exceptionSimpleDesc(e));
}
return null;
}