RocketMQ源码：broker 消息接收流程(写入commitLog)

清疚 2024-03-26 21:53 114阅读 0赞

从本文开始，我们来分析`rocketMq`消息接收、分发以及投递流程。

`RocketMq`消息处理整个流程如下：

![format_png][]

1.  消息接收：消息接收是指接收`producer`的消息，处理类是`SendMessageProcessor`，将消息写入到`commigLog`文件后，接收流程处理完毕；
2.  消息分发：`broker`处理消息分发的类是`ReputMessageService`，它会启动一个线程，不断地将`commitLong`分到到对应的`consumerQueue`，这一步操作会写两个文件：`consumerQueue`与`indexFile`，写入后，消息分发流程处理 完毕；
3.  消息投递：消息投递是指将消息发往`consumer`的流程，`consumer`会发起获取消息的请求，`broker`收到请求后，调用`PullMessageProcessor`类处理，从`consumerQueue`文件获取消息，返回给`consumer`后，投递流程处理完毕。

以上就是`rocketMq`处理消息的流程了，接下来我们就从源码来看相关流程的实现。

## 1. `remotingServer`的启动流程 ##

在正式分析接收与投递流程前，我们来了解下`remotingServer`的启动。

> `remotingServer`是一个netty服务，他开启了一个端口用来处理`producer`与`consumer`的网络请求。

`remotingServer`是在`BrokerController#start`中启动的，代码如下：

public void start() throws Exception {
            // 启动各组件
            ...
    
            if (this.remotingServer != null) {
                this.remotingServer.start();
            }
    
            ...
        }
    复制代码

继续查看`remotingServer`的启动流程，进入`NettyRemotingServer#start`方法：

public void start() {
        ...
    
        ServerBootstrap childHandler =
            this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                ...
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                            .addLast(defaultEventExecutorGroup, 
                                HANDSHAKE_HANDLER_NAME, handshakeHandler)
                            .addLast(defaultEventExecutorGroup,
                                encoder,
                                new NettyDecoder(),
                                new IdleStateHandler(0, 0, 
                                    nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                connectionManageHandler,
                                // 处理业务请求的handler
                                serverHandler
                            );
                    }
                });
    
        ...
    
    }
    复制代码

这就是一个标准的`netty`服务启动流程了，套路与`nameServer`的启动是一样的。关于`netty`的相关内容，这里我们仅关注`pipeline`上的`channelHandler`，在`netty`中，处理读写请求的操作为一个个`ChannelHandler`，`remotingServer`中处理读写请求的`ChanelHandler`为`NettyServerHandler`，代码如下：

@ChannelHandler.Sharable
    class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
            processMessageReceived(ctx, msg);
        }
    }
    复制代码

这块的操作与`nameServer`对外提供的服务极相似（就是同一个类），最终调用的是`NettyRemotingAbstract#processRequestCommand`方法：

public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
         // 根据 code 从 processorTable 获取 Pair
        final Pair<NettyRequestProcessor, ExecutorService> matched 
            = this.processorTable.get(cmd.getCode());
        // 找不到默认值    
        final Pair<NettyRequestProcessor, ExecutorService> pair =  
            null == matched ? this.defaultRequestProcessor : matched;
    
        ...
    
        // 从 pair 中拿到 Processor 进行处理
        NettyRequestProcessor processor = pair.getObject1();
        // 处理请求
        RemotingCommand response = processor.processRequest(ctx, cmd);
    
        ....
     }
    
    复制代码

如果进入源码去看，会发现这个方法非常长，这里省略了异步处理、异常处理及返回值构造等，仅列出了关键步骤：

1.  根据`code`从`processorTable`拿到对应的`Pair`
2.  从`Pair`里获取`Processor`

最终处理请求的就是`Processor`了。

## 2. `Processor`的注册 ##

从上面的分析中可知， `Processor`是处理消息的关键，它是从`processorTable`中获取的，这个`processorTable`是啥呢？

`processorTable`是`NettyRemotingAbstract`成员变量，里面的内容是`BrokerController`在初始化时（执行`BrokerController#initialize`方法）注册的。之前在分析`BrokerController`的初始化流程时，就提到过`Processor`的提供操作，这里再回顾下：

`BrokerController`的初始化方法`initialize`会调用 `BrokerController#registerProcessor`，`Processor`的注册操作就在这个方法里：

public class BrokerController {
    
        private final PullMessageProcessor pullMessageProcessor;
    
        /**
         * 构造方法
         */
        public BrokerController(...) {
            // 处理 consumer 拉消息请求的
            this.pullMessageProcessor = new PullMessageProcessor(this);
        }
    
        /**
         * 注册操作
         */
        public void registerProcessor() {
            // SendMessageProcessor
            SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
            sendProcessor.registerSendMessageHook(sendMessageHookList);
            sendProcessor.registerConsumeMessageHook(consumeMessageHookList);
            // 处理 Processor
            this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, 
                sendProcessor, this.sendMessageExecutor);
            this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, 
                sendProcessor, this.sendMessageExecutor);
            this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, 
                sendProcessor, this.sendMessageExecutor);
    
            // PullMessageProcessor
            this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, 
                this.pullMessageProcessor, this.pullMessageExecutor);
    
            // 省略其他许许多多的Processor注册    
            ...
    
        }
    
        ...
    复制代码

需要指明的是，`sendProcessor`用来处理`producer`请求过来的消息，`pullMessageProcessor`用来处理`consumer`拉取消息的请求。

## 3. 接收`producer`消息 ##

了解完`remotingServer`的启动与`Processor`的注册内容后，接下来我们就可以分析接收`producer`消息的流程了。

`producer`发送消息到`broker`时，发送的请求`code`为`SEND_MESSAGE`（[RocketMQ源码5-producer 同步发送和单向发送 第1.4小节][RocketMQ_5-producer _ _1.4]），根据上面的分析，当消息过来时，会使用`NettyServerHandler`这个`ChannelHandler`来处理，之后会调用到`NettyRemotingAbstract#processRequestCommand`方法。

在`NettyRemotingAbstract#processRequestCommand`方法中，会根据消息的`code`获取对应的`Processor`来处理，从`Processor`的注册流程来看，处理该`SEND_MESSAGE`的`Processor`为`SendMessageProcessor`，我们进入`SendMessageProcessor#processRequest`看看它的流程：

public RemotingCommand processRequest(ChannelHandlerContext ctx,
            RemotingCommand request) throws RemotingCommandException {
        RemotingCommand response = null;
        try {

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/6fde00e9127d4c219b0c238d54f62340.png
[RocketMQ_5-producer _ _1.4]: https://juejin.cn/post/7210412098812051515#heading-2