Rocketmq源码分析05：broker 消息接收流程

注：本系列源码分析基于RocketMq 4.8.0，gitee仓库链接：https://gitee.com/funcy/rocketmq.git.

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

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

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（这块内容在后面分析producer时再分析，暂时先当成一个结论吧），根据上面的分析，当消息过来时，会使用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 {
        // broker处理接收消息
        response = asyncProcessRequest(ctx, request).get();
    } catch (InterruptedException | ExecutionException e) {
        log.error("process SendMessage error, request : " + request.toString(), e);
    }
    return response;
}

没干啥事，一路跟下去，直接看普通消息的流程，进入SendMessageProcessor#asyncSendMessage方法：

private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, 
        RemotingCommand request, SendMessageContext mqtraceContext, 
        SendMessageRequestHeader requestHeader) {
    final RemotingCommand response = preSend(ctx, request, requestHeader);
    final SendMessageResponseHeader responseHeader 
        = (SendMessageResponseHeader)response.readCustomHeader();

    if (response.getCode() != -1) {
        return CompletableFuture.completedFuture(response);
    }

    final byte[] body = request.getBody();

    int queueIdInt = requestHeader.getQueueId();
    TopicConfig topicConfig = this.brokerController.getTopicConfigManager()
        .selectTopicConfig(requestHeader.getTopic());

    // 如果没指定队列，就随机指定一个队列
    if (queueIdInt < 0) {
        queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());
    }

    // 将消息包装为 MessageExtBrokerInner
    MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
    msgInner.setTopic(requestHeader.getTopic());
    msgInner.setQueueId(queueIdInt);

    // 省略处理 msgInner 的流程
    ...

    CompletableFuture<PutMessageResult> putMessageResult = null;
    Map<String, String> origProps = MessageDecoder
        .string2messageProperties(requestHeader.getProperties());
    String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
    // 发送事务消息
    if (transFlag != null && Boolean.parseBoolean(transFlag)) {
        ...
        // 发送事务消息
        putMessageResult = this.brokerController.getTransactionalMessageService()
            .asyncPrepareMessage(msgInner);
    } else {
        // 发送普通消息
        putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
    }
    return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, 
        responseHeader, mqtraceContext, ctx, queueIdInt);
}

这个方法是在准备消息的发送数据，所做的工作如下：

1. 如果没指定队列，就随机指定一个队列，一般情况下不会给消息指定队列的，但如果要发送顺序消息，就需要指定队列了，这点后面再分析。
2. 构造MessageExtBrokerInner对象，就是将producer上送的消息包装下，加上一些额外的信息，如消息标识msgId、发送时间、topic、queue等。
3. 发送消息，这里只是分为两类：事务消息与普通消息，这里我们主要关注普通消息，事务消息后面再分析。

进入普通消息的发送方法DefaultMessageStore#asyncPutMessage：

public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
    ...
    // 保存到 commitLog
    CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
    ...
}

继续进入CommitLog#asyncPutMessage方法：

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
    msg.setStoreTimestamp(System.currentTimeMillis());
    msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
    AppendMessageResult result = null;
    StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
    String topic = msg.getTopic();
    int queueId = msg.getQueueId();

    final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
    if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
            || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        // 延迟消息
        if (msg.getDelayTimeLevel() > 0) {
            // 延迟级别
            if (msg.getDelayTimeLevel() > this.defaultMessageStore
                    .getScheduleMessageService().getMaxDelayLevel()) {
                msg.setDelayTimeLevel(this.defaultMessageStore
                    .getScheduleMessageService().getMaxDelayLevel());
            }
            topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
            queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
            // 保存真正的 topic 与 queueId
            MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
            MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, 
                String.valueOf(msg.getQueueId()));
            msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
            // 换了一个topic与队列
            msg.setTopic(topic);
            msg.setQueueId(queueId);
        }
    }

    long elapsedTimeInLock = 0;
    MappedFile unlockMappedFile = null;
    MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

    putMessageLock.lock();
    try {
        long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
        this.beginTimeInLock = beginLockTimestamp;

        ...

        // 追加到文件中
        result = mappedFile.appendMessage(msg, this.appendMessageCallback);
        ...

        elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
        beginTimeInLock = 0;
    } finally {
        putMessageLock.unlock();
    }

    ...
}

在源码里，这个方法也是非常长，这里删减了大部分，只看关键点：

1. 如果发送的是延迟消息，先保存原始的topic与queueId，然后使用延迟队列专有的topic与queueId
2. 将消息写入到文件中

将消息写入到文件的操作是在MappedFile#appendMessage(...)方法中进行，关于这块就不过多分析了，我们直接看官方的描述(链接：https://github.com/apache/rocketmq/blob/master/docs/cn/design.md)：

rocketMq 消息存储架构图

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

(1) CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G ，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；

(2) ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；

(3) IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：HOME\store\index{fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

在上面的RocketMQ的消息存储整体架构图中可以看出，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。

正因为如此，Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后，可以等下一次消息拉取，同时服务端也支持长轮询模式，如果一个消息拉取请求未拉取到消息，Broker允许等待30s的时间，只要这段时间内有新消息到达，将直接返回给消费端。这里，RocketMQ的具体做法是，使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。

当消息写入commitlog文件后，producer发送消息的流程就结束了，接下来就是是消息的分发及消费流程了。

4. 总结

本文主要分析了 broker 接收producer消息的流程，流程如下：

1. 处理消息接收的底层服务为 netty，在BrokerController#start方法中启动
2. netty服务中，处理消息接收的channelHandler为NettyServerHandler，最终会调用SendMessageProcessor#processRequest来处理消息接收
3. 消息接收流程的最后，MappedFile#appendMessage(...)方法会将消息内容写入到commitLog文件中。

本文的分析就到这里了，下一篇我们继续分析commitLog文件的后续处理。

限于作者个人水平，文中难免有错误之处，欢迎指正！原创不易，商业转载请联系作者获得授权，非商业转载请注明出处。

本文首发于微信公众号 Java技术探秘，如果您喜欢本文，欢迎关注该公众号，让我们一起在技术的世界里探秘吧！