《RocketMQ源码分析》Broker是如何处理心跳的？

爱被打了一巴掌 2023-10-02 16:29 9阅读 0赞

## 一、Broker接收请求 ##

Broker作为服务端接收请求的流程如下图：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]  
接收到请求之后，我们着重看一下`NettyRemotingAbstract#processRequestCommand()`方法，看一下它是怎么处理客户端的请求的。

public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        
        // 根据Command的Code从processorTable中获取相应的事件处理器和线程池
        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
    
        // 找不到事件处理器，则使用默认的处理器
        final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
        final int opaque = cmd.getOpaque();
    
        if (pair != null) {
        
            Runnable run = new Runnable() {
        
                @Override
                public void run() {
        
                    try {
        
                        doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
                        final RemotingResponseCallback callback = new RemotingResponseCallback() {
        
                            @Override
                            public void callback(RemotingCommand response) {
        
                                doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
                                if (!cmd.isOnewayRPC()) {
        
                                    if (response != null) {
        
                                        response.setOpaque(opaque);
                                        response.markResponseType();
                                        try {
        
                                            ctx.writeAndFlush(response);
                                        } catch (Throwable e) {
        
                                            log.error("process request over, but response failed", e);
                                            log.error(cmd.toString());
                                            log.error(response.toString());
                                        }
                                    } else {
        
                                    }
                                }
                            }
                        };
                        if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
        
                            // 从 pair 中拿到 Processor事件处理器
                            AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor) pair.getObject1();
                            // 处理请求
                            processor.asyncProcessRequest(ctx, cmd, callback);
                        } else {
        
                            NettyRequestProcessor processor = pair.getObject1();
                            RemotingCommand response = processor.processRequest(ctx, cmd);
                            callback.callback(response);
                        }
                    } catch (Throwable e) {
        
                        log.error("process request exception", e);
                        log.error(cmd.toString());
    
                        if (!cmd.isOnewayRPC()) {
        
                            final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                                    RemotingHelper.exceptionSimpleDesc(e));
                            response.setOpaque(opaque);
                            ctx.writeAndFlush(response);
                        }
                    }
                }
            };
    
            if (pair.getObject1().rejectRequest()) {
        
                final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                        "[REJECTREQUEST]system busy, start flow control for a while");
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                return;
            }
    
            try {
        
                final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
                pair.getObject2().submit(requestTask);
            } catch (RejectedExecutionException e) {
        
                if ((System.currentTimeMillis() % 10000) == 0) {
        
                    log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                            + ", too many requests and system thread pool busy, RejectedExecutionException "
                            + pair.getObject2().toString()
                            + " request code: " + cmd.getCode());
                }
    
                if (!cmd.isOnewayRPC()) {
        
                    final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                            "[OVERLOAD]system busy, start flow control for a while");
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                }
            }
        } else {
        
            String error = " request type " + cmd.getCode() + " not supported";
            final RemotingCommand response =
                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
        }
    }

因为`ClientManageProcessor`是`AsyncNettyRequestProcessor`的子类，所以会走asyncProcessRequest()方法异步处理请求。  
![在这里插入图片描述][a78e4a973114469395932b1342ecf5f3.png]

在`AsyncNettyRequestProcessor#asyncProcessRequest()`中会调用子类的`processRequest()`方法实现：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
Broker处理心跳也就在`ClientManageProcessor#processRequest()`中：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]

## 二、Broker处理心跳 ##

上面提到了Broker处理心跳是在`ClientManageProcessor#heartBeat()`方法中处理的，下面我们来看一下`heartBeat()`方法：

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()
        );
    
        // 处理心跳包中的消费者信息
        for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
        
            // 获取Broker端的消费组订阅信息
            SubscriptionGroupConfig subscriptionGroupConfig =
                this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(
                    data.getGroupName());
            boolean isNotifyConsumerIdsChangedEnable = true;
            // 如果Broker端的消费组订阅信息不为空，说明当前可能要修改消费者订阅信息
            if (null != subscriptionGroupConfig) {
        
                // 是否通知到所有的消费者 订阅信息变更
                isNotifyConsumerIdsChangedEnable = subscriptionGroupConfig.isNotifyConsumerIdsChangedEnable();
                int topicSysFlag = 0;
                if (data.isUnitMode()) {
        
                    topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
                }
                // 消费失败后的，消息消费重试队列、名为：%RETRY%groupName
                String newTopic = MixAll.getRetryTopic(data.getGroupName());
                // 创建消息消费重试队列
                this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
                    newTopic,
                    subscriptionGroupConfig.getRetryQueueNums(),
                    PermName.PERM_WRITE | PermName.PERM_READ, topicSysFlag);
            }
    
            // 注册消费者订阅信息到Broker中，并判断订阅信息是否变更
            boolean changed = this.brokerController.getConsumerManager().registerConsumer(
                data.getGroupName(),
                clientChannelInfo,
                data.getConsumeType(),
                data.getMessageModel(),
                data.getConsumeFromWhere(),
                data.getSubscriptionDataSet(),
                isNotifyConsumerIdsChangedEnable
            );
    
            // 消费者信息发生变更，则打印日志记录
            if (changed) {
        
                log.info("registerConsumer info changed {} {}",
                    data.toString(),
                    RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                );
            }
        }
    
        // 处理心跳包中的生产者信息
        for (ProducerData data : heartbeatData.getProducerDataSet()) {
        
            // 直接注册producer，并把Producer的ClientChannelInfo保存下来、用于后面与Producer通信
            this.brokerController.getProducerManager().registerProducer(data.getGroupName(),
                clientChannelInfo);
        }
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

对于Producer的心跳信息处理非常简单，直接注册Producer、并把Producer的ClientChannelInfo保存下来、用于后面与Producer通信；

对于Consumer的心跳信息处理稍微复杂一点：

> 1、先判断Broker端中是否存在消费者的订阅信息：
> 
> > 1）如果存在，创建一个用于消息`重试消费的topic`（`%RETRY%`groupName）
> 
> 2、注册消费者订阅信息到Broker中；  
> 3、判断消费者订阅信息是否变更、消费者的clientChannel（所在机器IP+Port）是否变更；
> 
> > 只要有一个变更，就会通知当前消费组下的所有消费者实例 消费组中有信息变更。

最后我们看一下`ConusumerManager#registerConsumer()`注册消费者信息的源码：

public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
        ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
        final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {
        
    
        ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
        // 消费者初次上报订阅信息时
        if (null == consumerGroupInfo) {
        
            ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere);
            ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp);
            consumerGroupInfo = prev != null ? prev : tmp;
        }
    
        // 消费者的clientChannel（机器）是否变更
        boolean r1 =
            consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
                consumeFromWhere);
        // 消费者的订阅信息是否变更
        boolean r2 = consumerGroupInfo.updateSubscription(subList);
    
        // 若消费者的机器变更 或 订阅信息变更，则通知所有的消费者实例
        if (r1 || r2) {
        
            if (isNotifyConsumerIdsChangedEnable) {
        
                this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
            }
        }
    
        this.consumerIdsChangeListener.handle(ConsumerGroupEvent.REGISTER, group, subList);
    
        // 消费者信息发生变更，则返回true
        return r1 || r2;
    }

## 源码注释出处 ##

以上所有分析相关的源码注释请见GitHub中的release-4.8.0分支：[https://github.com/Saint9768/rocketmq/tree/rocketmq-all-4.8.0][https_github.com_Saint9768_rocketmq_tree_rocketmq-all-4.8.0]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_13_color_FFFFFF_t_70_g_se_x_16]

## 你们的每一个赞和关注都是博主创作的动力 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_10_color_FFFFFF_t_70_g_se_x_16_pic_left]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/132082f5218c443db777cfbae6ba406b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qWg546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[a78e4a973114469395932b1342ecf5f3.png]: https://img-blog.csdnimg.cn/a78e4a973114469395932b1342ecf5f3.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/94f9861acafe4efba708870d4d555bd5.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qWg546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/fa2e91b077024a32a2f431e9e9ba120a.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qWg546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[https_github.com_Saint9768_rocketmq_tree_rocketmq-all-4.8.0]: https://github.com/Saint9768/rocketmq/tree/rocketmq-all-4.8.0
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_13_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/cecee6393cc140c6aa35b06f42ac523b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_13,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qWg546W5p-S_size_10_color_FFFFFF_t_70_g_se_x_16_pic_left]: https://img-blog.csdnimg.cn/469409987fe34a758121afdce2bc9265.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qWg546W5p-S,size_10,color_FFFFFF,t_70,g_se,x_16#pic_left