关于rocketmq:RocketMQ源码broker-消息接收流程写入commitLog

从本文开始，咱们来剖析rocketMq音讯接管、散发以及投递流程。

RocketMq音讯解决整个流程如下：

音讯接管：音讯接管是指接管producer的音讯，解决类是SendMessageProcessor，将音讯写入到commigLog文件后，接管流程处理完毕；
音讯散发：broker解决音讯散发的类是ReputMessageService，它会启动一个线程，一直地将commitLong分到到对应的consumerQueue，这一步操作会写两个文件：consumerQueue与indexFile，写入后，音讯散发流程解决结束；
音讯投递：音讯投递是指将音讯发往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);

    ....
 }

如果进入源码去看，会发现这个办法十分长，这里省略了异步解决、异样解决及返回值结构等，仅列出了关键步骤：

依据code从processorTable拿到对应的Pair
从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大节），依据下面的剖析，当音讯过去时，会应用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);
}

这个办法是在筹备音讯的发送数据，所做的工作如下：

如果没指定队列，就随机指定一个队列，个别状况下不会给音讯指定队列的，但如果要发送程序音讯，就须要指定队列了，这点前面再剖析。
结构MessageExtBrokerInner对象，就是将producer上送的音讯包装下，加上一些额定的信息，如音讯标识msgId、发送工夫、topic、queue等。
发送音讯，这里只是分为两类：事务音讯与一般音讯，这里咱们次要关注一般音讯，事务音讯前面再剖析。

进入一般音讯的发送办法DefaultMessageStore#asyncPutMessage：

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

3.1 commitLog写入原理

一个broker逻辑上对应着一个commitLog，你能够把它看作一个大文件，而后这个broker收到的所有音讯都写到这个外面，然而物理上ROCKET_HOME/commitlog/00000000000000000000这个门路存储的，它是由若干个文件组成的，每个文件默认大小是1G，而后每个文件都对应这个一个MappedFile，00000000000000000000 就是第一个MappedFile对应的物理文件，每个文件的文件名就是在commitLog外面的一个其实offset，第二个文件名就是00000000001073741824，也就是上一个MappedFile文件起始offset加上每个文件的大小，这个MappedFile就是RocketMQ的黑科技，应用了内存映射技术来进步文件的访问速度与写入速度，而后都是采纳追加写的形式进步写入速度。

咱们间接看官网的形容(链接：github.com/apache/rock…)：

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（索引文件）数据。

3.2 CommitLog#asyncPutMessage

持续进入CommitLog#asyncPutMessage办法，这个办法有点长，咱们分局部：

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
    // Set the storage time 设置存储工夫
    msg.setStoreTimestamp(System.currentTimeMillis());
    // Set the message body BODY CRC (consider the most appropriate setting
    // on the client)
    // 设置crc
    msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
    // Back to Results
    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) {
        // Delay Delivery 延时音讯的解决
        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());

            // Backup real 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()));

            msg.setTopic(topic);
            msg.setQueueId(queueId);
        }
    }
    ...

这一部分其实就是从msg中获取一些信息，判断解决一下这个延时音讯。

long elapsedTimeInLock = 0;
MappedFile unlockMappedFile = null;
// 获取最初一个 MappedFile
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

// 获取写入锁
putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
try {
    long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
    // 开始在锁里的工夫
    this.beginTimeInLock = beginLockTimestamp;

    // Here settings are stored timestamp, in order to ensure an orderly
    // global
    // 设置写入的工夫戳，确保它是有序的，
    msg.setStoreTimestamp(beginLockTimestamp);

    // 判断MappedFile 是否是 null 或者是否是满了
    if (null == mappedFile || mappedFile.isFull()) {
        mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
    }
    if (null == mappedFile) {
        log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
        beginTimeInLock = 0;
        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));
    }

    // todo 往mappedFile追加音讯
    result = mappedFile.appendMessage(msg, this.appendMessageCallback);

    ...

这一部分就比拟重要了，首先是从mappedFileQueue中获取最初一个MappedFile，这个就是拿汇合最初一个元素，因为都是有序的，最初一个元素就是最初一个MappedFile，接着就是获取锁了，这个锁也是比拟有考究的，能够设置应用ReentrantLock 也能够设置应用cas，默认是应用cas，接着就是设置beginTimeInLock这个变量了，这个变量咱们在判断os page cache忙碌的时候说过，就是获取到锁的一个工夫戳，在开释锁之前会重置成0，接着就是判断mappedFile是不是null或者是不是满了，如果是的话就要新建一个了。

接着就是最最最重要的了往mappedFile中追加音讯，

mappedFile.appendMessage

/**
 * 将音讯追加到MappedFile文件中
 */
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
    assert messageExt != null;
    assert cb != null;

    // 获取MappedFile以后文件写指针
    int currentPos = this.wrotePosition.get();

    // 如果currentPos小于文件大小
    if (currentPos < this.fileSize) {
        ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
        byteBuffer.position(currentPos);
        AppendMessageResult result;
        // 单个音讯
        if (messageExt instanceof MessageExtBrokerInner) {
            // todo
            result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
        // 批量音讯
        } else if (messageExt instanceof MessageExtBatch) {
            result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
        } else {
            return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
        }
        this.wrotePosition.addAndGet(result.getWroteBytes());
        this.storeTimestamp = result.getStoreTimestamp();
        return result;
    }
    // 如果currentPos大于或等于文件大小，表明文件已写满，抛出异样
    log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}

这里首先获取了一下这个mappedFile写到哪个地位了，它这个地位是从0开始的，而后判断一下以后地位与文件大小做比照，要是大于的就超了文件大小了，接着是获取writerbuffer因为这里是没有开启transientStorePool的，所以它是个空的，就会应用mmapedByteBuffer，接着就是调用回调的doAppend追加音讯了，咱们看下它的参数，第一个是开始offset，这个offset是commitlog的一个offset，举个例子，第一个MappedFile的开始offset是0，而后一个MappedFile 的大小是1g，而后第二个MappedFile就得从1073741824（1g）开始了，第二个参数是bytebuffer，这个不必多说，第三个是这个MappedFile还空多少字节没用，第四个就是音讯了。

咱们来看下这个doAppend办法，这个也有点长，咱们须要离开看下：

/**
 * // 只是将音讯追加到内存中
 * @param fileFromOffset 文件的第一个偏移量（就是MappedFile是从哪个中央开始的）
 */
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
    final MessageExtBrokerInner msgInner) {
    // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>

    // PHY OFFSET
    long wroteOffset = fileFromOffset + byteBuffer.position();

    int sysflag = msgInner.getSysFlag();

    int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
    int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
    ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
    ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);

    this.resetByteBuffer(storeHostHolder, storeHostLength);
    // 创立全局惟一音讯id
    String msgId;
    if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
        msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
    } else {
        msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
    }

    // Record ConsumeQueue information
    keyBuilder.setLength(0);
    keyBuilder.append(msgInner.getTopic());
    keyBuilder.append('-');
    keyBuilder.append(msgInner.getQueueId());
    String key = keyBuilder.toString();
    // 获取该音讯在音讯队列的物理偏移量
    Long queueOffset = CommitLog.this.topicQueueTable.get(key);
    if (null == queueOffset) {
        queueOffset = 0L;
        CommitLog.this.topicQueueTable.put(key, queueOffset);
    }

    // Transaction messages that require special handling
    final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
    switch (tranType) {
        // Prepared and Rollback message is not consumed, will not enter the
        // consumer queue
        case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
        case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
            queueOffset = 0L;
            break;
        case MessageSysFlag.TRANSACTION_NOT_TYPE:
        case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
        default:
            break;
    }

这一部分次要就是计算了一下这个音讯写在commitlog中的一个offset，接着就是生成一个msgId，而后依据topic 与queueId从缓存中获取了一下这个queueId对应的一个queue的offset，这个其实就是增加一个音讯加1，而后就是事务的货色了，如果有事务，而后还在筹备阶段或者回滚阶段，就将queue offset 设置成0，再往下其实就是解决音讯，而后写到buffer中了。

/**
 * Serialize message
 */
final byte[] propertiesData =
    msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);

final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;

if (propertiesLength > Short.MAX_VALUE) {
    log.warn("putMessage message properties length too long. length={}", propertiesData.length);
    return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
}

final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
final int topicLength = topicData.length;

final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;

// todo 计算音讯总长度
final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);

// Exceeds the maximum message  
if (msgLen > this.maxMessageSize) {  // 最大音讯长度 65536
    CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
        + ", maxMessageSize: " + this.maxMessageSize);
    return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
}
...

这里首先是获取了一下音讯外面的properties，将它转成字节数组，计算了一下长度，接着就是将topic转成字节数据，计算了一下长度，获取了一下body的长度，就是你往Message塞得内容长度，重点来，计算音讯的总长度，而后判断一下长度是否超长。其中calMsgLength如下：

    // 计算音讯长度
    protected static int calMsgLength(int sysFlag, int bodyLength, int topicLength, int propertiesLength) {
        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
        int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
        final int msgLen = 4 //TOTALSIZE 音讯条目总长度，4字节
            + 4 //MAGICCODE 魔数，4字节。固定值0xdaa320a7
            + 4 //BODYCRC 音讯体的crc校验码，4字节
            + 4 //QUEUEID 音讯生产队列ID，4字节
            + 4 //FLAG 音讯标记，RocketMQ对其不做解决，供应用程序应用， 默认4字节
            + 8 //QUEUEOFFSET 音讯在ConsumeQuene文件中的物理偏移量，8字节。
            + 8 //PHYSICALOFFSET 音讯在CommitLog文件中的物理偏移量，8字节
            + 4 //SYSFLAG 音讯零碎标记，例如是否压缩、是否是事务音讯 等，4字节
            + 8 //BORNTIMESTAMP 音讯生产者调用音讯发送API的工夫戳，8字 节
            + bornhostLength //BORNHOST 音讯发送者IP、端口号，8字节
            + 8 //STORETIMESTAMP 音讯存储工夫戳，8字节
            + storehostAddressLength //STOREHOSTADDRESS  Broker服务器IP+端口号，8字节
            + 4 //RECONSUMETIMES 音讯重试次数，4字节
            + 8 //Prepared Transaction Offset 事务音讯的物理偏移量，8 字节。
            + 4  // 音讯体长度，4字节
            + (bodyLength > 0 ? bodyLength : 0) // BODY  音讯体内容，长度为bodyLenth中存储的值
            + 1 // 主题存储长度，1字节，示意主题名称不能超过255个字符。
            + topicLength //TOPIC 主题，长度为TopicLength中存储的值
            + 2  // 音讯属性长度，2字节，示意音讯属性长 度不能超过65536个字符。
            + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength  音讯属性，长度为PropertiesLength中存储的 值
            + 0;
        return msgLen;
    }

持续：

...
// todo 音讯长度+END_FILE_MIN_BLANK_LENGTH 大于commitLog的闲暇空间，则返回END_OF_FILE
if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
    this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
    // 1 TOTALSIZE  4字节存储以后文件的残余空间
    this.msgStoreItemMemory.putInt(maxBlank);
    // 2 MAGICCODE 4字节存储魔数
    this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
    // 3 The remaining space may be any value
    // Here the length of the specially set maxBlank
    final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
    byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
    return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
        queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
}
...

判断剩下的空间能不能放开，如果放不开的话，就塞上一个完结的货色，8个字节是正经的，剩下的随便，而后返回文件满了的状态。

...
// Initialization of storage space
// 初始化存储空间
this.resetByteBuffer(msgStoreItemMemory, msgLen);
// 1 TOTALSIZE
this.msgStoreItemMemory.putInt(msgLen);
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
// 3 BODYCRC
this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
// 4 QUEUEID
this.msgStoreItemMemory.putInt(msgInner.getQueueId());
// 5 FLAG
this.msgStoreItemMemory.putInt(msgInner.getFlag());
// 6 QUEUEOFFSET
this.msgStoreItemMemory.putLong(queueOffset);
// 7 PHYSICALOFFSET
this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
// 8 SYSFLAG
this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
// 9 BORNTIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
// 10 BORNHOST
this.resetByteBuffer(bornHostHolder, bornHostLength);
this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
// 11 STORETIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
// 12 STOREHOSTADDRESS
this.resetByteBuffer(storeHostHolder, storeHostLength);
this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
// 13 RECONSUMETIMES
this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
// 14 Prepared Transaction Offset
this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
// 15 BODY
this.msgStoreItemMemory.putInt(bodyLength);
if (bodyLength > 0)
    this.msgStoreItemMemory.put(msgInner.getBody());
// 16 TOPIC
this.msgStoreItemMemory.put((byte) topicLength);
this.msgStoreItemMemory.put(topicData);
// 17 PROPERTIES
this.msgStoreItemMemory.putShort((short) propertiesLength);
if (propertiesLength > 0)
    this.msgStoreItemMemory.put(propertiesData);

final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
// Write messages to the queue buffer
byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
...

这个就是封装音讯了，最初将音讯放到byteBuffer中。

...
// 创立AppendMessageResult
    AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
        msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);

    switch (tranType) {
        case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
        case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
            break;
        case MessageSysFlag.TRANSACTION_NOT_TYPE:
        case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
            // The next update ConsumeQueue information
            // 更新音讯队列的逻辑偏移量
            CommitLog.this.topicQueueTable.put(key, ++queueOffset);
            break;
        default:
            break;
    }
    return result;
}

最初就是封装追加音讯的后果是put_ok，而后更新queue offset ，其实就是+1。

接下来咱们回过头来看下appendMessagesInner的后半局部，

...
this.wrotePosition.addAndGet(result.getWroteBytes());
this.storeTimestamp = result.getStoreTimestamp();
return result;

这里其实就是更新了一下这个MappedFile 写到哪个中央了，更新了下写入工夫。

回到commitLog的putMessage办法:

...
    // todo 往mappedFile追加音讯
    result = mappedFile.appendMessage(msg, this.appendMessageCallback);
    switch (result.getStatus()) {
        case PUT_OK:
            break;
        case END_OF_FILE:
            unlockMappedFile = mappedFile;
            // Create a new file, re-write the message
            mappedFile = this.mappedFileQueue.getLastMappedFile(0);
            if (null == mappedFile) {
                // XXX: warn and notify me
                log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                beginTimeInLock = 0;
                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));
            }
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            break;
        case MESSAGE_SIZE_EXCEEDED:
        case PROPERTIES_SIZE_EXCEEDED:
            beginTimeInLock = 0;
            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
        case UNKNOWN_ERROR:
            beginTimeInLock = 0;
            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
        default:
            beginTimeInLock = 0;
            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
    }

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

这里追加实现了，就须要判断追加状态了，如果是那种MappedFile放不开音讯的状况，它会从新获取一个MappedFile，而后从新追加，在开释锁之前，它还会将beginTimeInLock这个字段重置为0；

...
if (elapsedTimeInLock > 500) {
        log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
    }

    if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
        this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
    }

    PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);

    // Statistics
    storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
    storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());

    // todo 音讯首先进入pagecache，而后执行刷盘操作，
    CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
    // todo 接着调用submitReplicaRequest办法将音讯提交到HaService,进行数据复制
    CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);

    // todo 这里应用了ComplateFuture的thenCombine办法，将刷盘、复制当成一
    // todo 个联结工作执行，这里设置音讯追加的最终状态
    return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
        if (flushStatus != PutMessageStatus.PUT_OK) {
            putMessageResult.setPutMessageStatus(flushStatus);
        }
        if (replicaStatus != PutMessageStatus.PUT_OK) {
            putMessageResult.setPutMessageStatus(replicaStatus);
            if (replicaStatus == PutMessageStatus.FLUSH_SLAVE_TIMEOUT) {
                log.error("do sync transfer other node, wait return, but failed, topic: {} tags: {} client address: {}",
                        msg.getTopic(), msg.getTags(), msg.getBornHostNameString());
            }
        }
        return putMessageResult;
    });
}

判断了一下耗时，如果是大于500ms的话，打印正告，封装put音讯的后果，统计store，能够看到前面调用了2个办法，一个是刷盘的，一个是同步音讯的，咱们这里要看下这个刷盘动作:

public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {
    // Synchronization flush 同步刷盘
    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
        final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
        if (messageExt.isWaitStoreMsgOK()) {
            // 构建GroupCommitRequest同步工作并提交到GroupCommitRequest
            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
                    this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
            // 刷盘申请
            service.putRequest(request);
            return request.future();
        } else {
            service.wakeup();
            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        }
    }
    // Asynchronous flush  异步刷盘 这个就是靠os
    else {
        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            flushCommitLogService.wakeup();
        } else  {
            commitLogService.wakeup();
        }
        return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
    }
}

如果broker配置的SYNC_FLUSH 并且是个同步音讯，这个时候就会创立一个刷盘申请，而后提交刷盘申请，这个时候会等着刷盘实现，默认就是5s。

接着就是到存储器的putMessage办法的后半局部了:

...
// todo 存储音讯
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);

putResultFuture.thenAccept((result) -> {
    long elapsedTime = this.getSystemClock().now() - beginTime;
    if (elapsedTime > 500) {
        log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
    }
    // 记录状态
    this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);

    if (null == result || !result.isOk()) {
        // 记录状态
        this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
    }
});

return putResultFuture;

commitlog存入音讯之后，咱们这块也就算是实现了，最初就是回到那个processor，而后将put后果写入对应的channel给返回去，通知音讯生产者音讯写入后果。音讯存储其实就是找对应的MappedFile，依照肯定的格局往文件外面写入，须要留神的是内存映射文件。

这里附一张音讯存储字段存储程序与字段长度的图：

4. 总结

本文次要剖析了 broker 接管producer音讯的流程，流程如下：

解决音讯接管的底层服务为 netty，在BrokerController#start办法中启动
netty服务中，解决音讯接管的channelHandler为NettyServerHandler，最终会调用SendMessageProcessor#processRequest来解决音讯接管
音讯接管流程的最初，MappedFile#appendMessage(…)办法会将音讯内容写入到commitLog文件中。

关于rocketmq:RocketMQ源码broker-消息接收流程写入commitLog

1. remotingServer的启动流程

2. Processor的注册

3. 接管producer音讯

3.1 commitLog写入原理

3.2 CommitLog#asyncPutMessage

mappedFile.appendMessage

4. 总结

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

关于rocketmq:RocketMQ源码broker-消息接收流程写入commitLog

1. remotingServer的启动流程

2. Processor的注册

3. 接管producer音讯

3.1 commitLog写入原理

3.2 CommitLog#asyncPutMessage

mappedFile.appendMessage

4. 总结

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复