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

java技术探秘

共 21533字,需浏览 44分钟

 ·

2021-04-22 21:34

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

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

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

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

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

1. remotingServer的启动流程

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

remotingServer是一个netty服务,他开启了一个端口用来处理producerconsumer的网络请求。

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(00
                                nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                            connectionManageHandler,
                            // 处理业务请求的handler
                            serverHandler
                        );
                }
            });

    ...

}

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

 @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. 根据codeprocessorTable拿到对应的Pair
  2. Pair里获取Processor

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

2. Processor的注册

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

processorTableNettyRemotingAbstract成员变量,里面的内容是BrokerController在初始化时(执行BrokerController#initialize方法)注册的。之前在分析BrokerController的初始化流程时,就提到过Processor的提供操作,这里再回顾下:

BrokerController的初始化方法initialize会调用 BrokerController#registerProcessorProcessor的注册操作就在这个方法里:

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时,发送的请求codeSEND_MESSAGE(这块内容在后面分析producer时再分析,暂时先当成一个结论吧),根据上面的分析,当消息过来时,会使用NettyServerHandler这个ChannelHandler来处理,之后会调用到NettyRemotingAbstract#processRequestCommand方法。

NettyRemotingAbstract#processRequestCommand方法中,会根据消息的code获取对应的Processor来处理,从Processor的注册流程来看,处理该SEND_MESSAGEProcessorSendMessageProcessor,我们进入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、发送时间、topicqueue等。
  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. 如果发送的是延迟消息,先保存原始的topicqueueId,然后使用延迟队列专有的topicqueueId
  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和消息TagHashCode值。consumequeue文件可以看成是基于topiccommitlog索引文件,故consumequeue文件夹的组织方式如下:topic/queue/file三层组织结构,具体存储路径为:$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计,每一个条目共20个字节,分别为8字节的commitlog物理偏移量4字节的消息长度8字节tag hashcode,单个文件由30W个条目组成,可以像数组一样随机访问每一个条目,每个ConsumeQueue文件大小约5.72M;

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

在上面的RocketMQ的消息存储整体架构图中可以看出,RocketMQ采用的是混合型的存储结构,即为Broker单个实例下所有的队列共用一个日志数据文件(即为CommitLog)来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对ProducerConsumer分别采用了数据和索引部分相分离的存储结构,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服务中,处理消息接收的channelHandlerNettyServerHandler,最终会调用SendMessageProcessor#processRequest来处理消息接收
  3. 消息接收流程的最后,MappedFile#appendMessage(...)方法会将消息内容写入到commitLog文件中。

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


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

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

- END -


浏览 39
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报