一、简介

RocketMQ 是阿里巴巴开源的分布式消息中间件，它借鉴了 Kafka 实现，反对音讯订阅与公布、程序音讯、事务音讯、定时音讯、音讯回溯、死信队列等性能。RocketMQ 架构上次要分为四局部，如下图所示：

Producer：音讯生产者，反对分布式集群形式部署。
Consumer：音讯消费者，反对分布式集群形式部署。
NameServer：名字服务，是一个非常简单的 Topic 路由注册核心，反对 Broker 的动静注册与发现，Producer 和 Consumer 通过 NameServer 动静感知 Broker 的路由信息。
Broker：Broker 次要负责音讯的存储、转发和查问。

本文基于 Apache RocketMQ 4.9.1 版本分析 Broker 中的音讯存储模块是如何设计的。

二、存储架构

RocketMQ 的音讯文件门路如图所示。

CommitLog

音讯主体以及元数据的存储主体，存储 Producer 端写入的音讯主体内容，音讯内容不是定长的。单个文件大小默认1G，文件名长度为 20 位，右边补零，残余为起始偏移量，比方 00000000000000000000 代表了第一个文件，起始偏移量为 0，文件大小为 1G=1073741824；当第一个文件写满了，第二个文件为 00000000001073741824，起始偏移量为 1073741824，以此类推。

ConsumeQueue

音讯生产队列，Consumequeue 文件能够看成是基于 CommitLog 的索引文件。ConsumeQueue 文件采取定长设计，每一个条目共 20 个字节，别离为 8 字节的 CommitLog 物理偏移量、4 字节的音讯长度、8 字节 tag hashcode，单个文件由 30W 个条目组成，能够像数组一样随机拜访每一个条目，每个 ConsumeQueue 文件大小约 5.72M。

IndexFile

索引文件，提供了一种能够通过 key 或工夫区间来查问音讯的办法。单个 IndexFile 文件大小约为 400M，一个 IndexFile 能够保留 2000W 个索引，IndexFile 的底层存储设计相似 JDK 的 HashMap 数据结构。

其余文件：包含 config 文件夹，寄存运行时配置信息；abort 文件，阐明 Broker 是否失常敞开；checkpoint 文件，存储 Commitlog、ConsumeQueue、Index 文件最初一次刷盘工夫戳。这些不在本文探讨的范畴。

同 Kafka 相比，Kafka 每个 Topic 的每个 partition 对应一个文件，程序写入，定时刷盘。但一旦单个 Broker 的 Topic 过多，程序写将进化为随机写。而 RocketMQ 单个 Broker 所有 Topic 在同一个 CommitLog 中程序写，是可能保障严格程序写。RocketMQ 读取音讯须要从 ConsumeQueue 中拿到音讯理论物理偏移再去 CommitLog 读取音讯内容，会造成随机读取。

2.1 Page Cache 和 mmap

在正式介绍 Broker 音讯存储模块实现前，先阐明下 Page Cache 和 mmap 这两个概念。

Page Cache 是 OS 对文件的缓存，用于减速对文件的读写。一般来说，程序对文件进行程序读写的速度简直靠近于内存的读写速度，次要起因就是因为 OS 应用 Page Cache 机制对读写访问操作进行了性能优化，将一部分的内存用作 Page Cache。对于数据的写入，OS 会先写入至 Cache 内，随后通过异步的形式由 pdflush 内核线程将 Cache 内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时呈现未命中 Page Cache 的状况，OS 从物理磁盘上拜访读取文件的同时，会程序对其余相邻块的数据文件进行预读取。

mmap 是将磁盘上的物理文件间接映射到用户态的内存地址中，缩小了传统 IO 将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销。Java NIO 中的 FileChannel 提供了 map() 办法能够实现 mmap。FileChannel (文件通道)和 mmap (内存映射) 读写性能比拟能够参照这篇文章。

2.2 Broker 模块

下图是 Broker 存储架构图，展现了 Broker 模块从收到音讯到返回响应业务流转过程。

业务接入层：RocketMQ 基于 Netty 的 Reactor 多线程模型实现了底层通信。Reactor 主线程池 eventLoopGroupBoss 负责创立 TCP 连贯，默认只有一个线程。连贯建设后，再丢给 Reactor 子线程池 eventLoopGroupSelector 进行读写事件的解决。

defaultEventExecutorGroup 负责 SSL 验证、编解码、闲暇查看、网络连接治理。而后依据 RomotingCommand 的业务申请码 code 去 processorTable 这个本地缓存变量中找到对应的 processor，封装成 task 工作后，提交给对应的业务 processor 解决线程池来执行。Broker 模块通过这四级线程池晋升零碎吞吐量。

业务解决层：解决各种通过 RPC 调用过去的业务申请，其中：

SendMessageProcessor 负责解决 Producer 发送音讯的申请；
PullMessageProcessor 负责解决 Consumer 生产音讯的申请；
QueryMessageProcessor 负责解决依照音讯 Key 等查问音讯的申请。

存储逻辑层：DefaultMessageStore 是 RocketMQ 的存储逻辑外围类，提供音讯存储、读取、删除等能力。

文件映射层：把 Commitlog、ConsumeQueue、IndexFile 文件映射为存储对象 MappedFile。

数据传输层：反对基于 mmap 内存映射进行读写音讯，同时也反对基于 mmap 进行读取音讯、堆外内存写入音讯的形式进行读写音讯。

上面章节将从源码角度来分析 RocketMQ 是如何实现高性能存储。

三、音讯写入

以单个音讯生产为例，音讯写入时序逻辑如下图，业务逻辑如上文 Broker 存储架构所示在各层之间进行流转。

最底层音讯写入外围代码在 CommitLog 的 asyncPutMessage 办法中，次要分为获取 MappedFile、往缓冲区写音讯、提交刷盘申请三步。须要留神的是在这三步前后有自旋锁或 ReentrantLock 的加锁、开释锁，保障单个 Broker 写音讯是串行的。

//org.apache.rocketmq.store.CommitLog::asyncPutMessagepublic CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {        ...        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config        try {            //获取最新的 MappedFile            MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();            ...            //向缓冲区写音讯            result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);            ...            //提交刷盘申请            CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);            ...        } finally {            putMessageLock.unlock();        }        ...    }

上面介绍这三步具体做了什么事件。

3.1 MappedFile 初始化

在 Broker 初始化时会启动治理 MappedFile 创立的 AllocateMappedFileService 异步线程。音讯解决线程和 AllocateMappedFileService 线程通过队列 requestQueue 关联。

音讯写入时调用 AllocateMappedFileService 的 putRequestAndReturnMappedFile 办法往 requestQueue 放入提交创立 MappedFile 申请，这边会同时构建两个 AllocateRequest 放入队列。

AllocateMappedFileService 线程循环从 requestQueue 获取 AllocateRequest 来创立 MappedFile。音讯解决线程通过 CountDownLatch 期待获取第一个 MappedFile 创立胜利就返回。

当音讯解决线程须要再次创立 MappedFile 时，此时能够间接获取之前已预创立的 MappedFile。这样通过预创立 MappedFile ，缩小文件创建等待时间。

//org.apache.rocketmq.store.AllocateMappedFileService::putRequestAndReturnMappedFilepublic MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {    //申请创立 MappedFile    AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);    boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;    ...    //申请事后创立下一个 MappedFile    AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);    boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;    ...    //获取本次创立 MappedFile    AllocateRequest result = this.requestTable.get(nextFilePath);    ...} //org.apache.rocketmq.store.AllocateMappedFileService::runpublic void run() {    ..    while (!this.isStopped() && this.mmapOperation()) {    }    ...} //org.apache.rocketmq.store.AllocateMappedFileService::mmapOperationprivate boolean mmapOperation() {    ...    //从队列获取 AllocateRequest    req = this.requestQueue.take();    ...    //判断是否开启堆外内存池    if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {        //开启堆外内存的 MappedFile        mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();        mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());    } else {        //一般 MappedFile        mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());    }    ...    //MappedFile 预热    if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig()        .getMappedFileSizeCommitLog()        &&        this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {        mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),            this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());    }    req.setMappedFile(mappedFile);    ...}

每次新建一般 MappedFile 申请，都会创立 mappedByteBuffer，上面代码展现了 Java mmap 是如何实现的。

//org.apache.rocketmq.store.MappedFile::initprivate void init(final String fileName, final int fileSize) throws IOException {    ...    this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();    this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);    ...}

如果开启堆外内存，即 transientStorePoolEnable = true 时，mappedByteBuffer 只是用来读音讯，堆外内存用来写音讯，从而实现对于音讯的读写拆散。堆外内存对象不是每次新建 MappedFile 都须要创立，而是系统启动时依据堆外内存池大小就初始化好了。每个堆外内存 DirectByteBuffer 都与 CommitLog 文件大小雷同，通过锁定住该堆外内存，确保不会被置换到虚拟内存中去。

//org.apache.rocketmq.store.TransientStorePoolpublic void init() {    for (int i = 0; i < poolSize; i++) {        //调配与 CommitLog 文件大小雷同的堆外内存        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);        final long address = ((DirectBuffer) byteBuffer).address();        Pointer pointer = new Pointer(address);        //锁定堆外内存，确保不会被置换到虚拟内存中去        LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));        availableBuffers.offer(byteBuffer);    }}

下面的 mmapOperation 办法中有段 MappedFile 预热逻辑。为什么须要文件预热呢？文件预热怎么做的呢？

因为通过 mmap 映射，只是建设了过程虚拟内存地址与物理内存地址之间的映射关系，并没有将 Page Cache 加载至内存。读写数据时如果没有命中写 Page Cache 则产生缺页中断，从磁盘从新加载数据至内存，这样会影响读写性能。为了避免缺页异样，阻止操作系统将相干的内存页调度到替换空间（swap space），RocketMQ 通过对文件预热，文件预热实现如下。

//org.apache.rocketmq.store.MappedFile::warmMappedFilepublic void warmMappedFile(FlushDiskType type, int pages) {        ByteBuffer byteBuffer = this.mappedByteBuffer.slice();        int flush = 0;        //通过写入 1G 的字节 0 来让操作系统调配物理内存空间，如果没有填充值，操作系统不会理论调配物理内存，避免在写入音讯时产生缺页异样        for (int i = 0, j = 0; i < this.fileSize; i += MappedFile.OS_PAGE_SIZE, j++) {            byteBuffer.put(i, (byte) 0);            // force flush when flush disk type is sync            if (type == FlushDiskType.SYNC_FLUSH) {                if ((i / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE) >= pages) {                    flush = i;                    mappedByteBuffer.force();                }            }             //prevent gc            if (j % 1000 == 0) {                Thread.sleep(0);            }        }         //force flush when prepare load finished        if (type == FlushDiskType.SYNC_FLUSH) {            mappedByteBuffer.force();        }        ...        this.mlock();} //org.apache.rocketmq.store.MappedFile::mlockpublic void mlock() {    final long beginTime = System.currentTimeMillis();    final long address = ((DirectBuffer) (this.mappedByteBuffer)).address();    Pointer pointer = new Pointer(address);     //通过零碎调用 mlock 锁定该文件的 Page Cache，避免其被替换到 swap 空间    int ret = LibC.INSTANCE.mlock(pointer, new NativeLong(this.fileSize));     //通过零碎调用 madvise 给操作系统倡议，阐明该文件在不久的未来要被拜访    int ret = LibC.INSTANCE.madvise(pointer, new NativeLong(this.fileSize), LibC.MADV_WILLNEED);}

综上所述，RocketMQ 每次都预创立一个文件来缩小文件创建提早，通过文件预热防止了读写时缺页异样。

3.2 音讯写入

3.2.1 写入 CommitLog

CommitLog 中每条音讯存储的逻辑视图如下图所示， TOTALSIZE 是整个音讯占用存储空间大小。

上面表格阐明下每条音讯蕴含哪些字段，以及这些字段占用空间大小和字段简介。

音讯的写入是调用MappedFile 的 appendMessagesInner办法。

//org.apache.rocketmq.store.MappedFile::appendMessagesInnerpublic AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb,        PutMessageContext putMessageContext) {    //判断应用 DirectBuffer 还是 MappedByteBuffer 进行写操作    ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();    ..    byteBuffer.position(currentPos);    AppendMessageResult result  = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,                    (MessageExtBrokerInner) messageExt, putMessageContext);    ..    return result;} //org.apache.rocketmq.store.CommitLog::doAppendpublic AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,    final MessageExtBrokerInner msgInner, PutMessageContext putMessageContext) {    ...    ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();    ...    //这边只是将音讯写入缓冲区，还未理论刷盘    byteBuffer.put(preEncodeBuffer);    msgInner.setEncodedBuff(null);    ...    return result;}

至此，音讯最终写入 ByteBuffer，还没有长久到磁盘，具体何时长久化，下一大节会具体讲刷盘机制。这边有个疑难 ConsumeQueue 和 IndexFile 是怎么写入的？

答案是在存储架构图中存储逻辑层的 ReputMessageService。MessageStore 在初始化的时候，会启动一个 ReputMessageService 异步线程，它启动后便会在循环中一直调用 doReput 办法，用来告诉 ConsumeQueue 和 IndexFile 进行更新。ConsumeQueue 和 IndexFile 之所以能够异步更新是因为 CommitLog 中保留了复原 ConsumeQueue 和 IndexFile 所需队列和 Topic 等信息，即便 Broker 服务异样宕机，Broker 重启后能够依据 CommitLog 复原 ConsumeQueue 和IndexFile。

//org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService::runpublic void run() {    ...    while (!this.isStopped()) {        Thread.sleep(1);         this.doReput();    }    ...} //org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService::doReputprivate void doReput() {    ...    //获取CommitLog中存储的新音讯    DispatchRequest dispatchRequest =        DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);    int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize();     if (dispatchRequest.isSuccess()) {        if (size > 0) {            //如果有新音讯，则别离调用 CommitLogDispatcherBuildConsumeQueue、CommitLogDispatcherBuildIndex 进行构建 ConsumeQueue 和 IndexFile            DefaultMessageStore.this.doDispatch(dispatchRequest);    }    ...}

3.2.2 写入 ConsumeQueue

如下图所示，ConsumeQueue 每一条记录共 20 个字节，别离为 8 字节的 CommitLog 物理偏移量、4 字节的音讯长度、8字节 tag hashcode。

ConsumeQueue 记录长久化逻辑如下。

//org.apache.rocketmq.store.ConsumeQueue::putMessagePositionInfoprivate boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,    final long cqOffset) {    ...    this.byteBufferIndex.flip();    this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);    this.byteBufferIndex.putLong(offset);    this.byteBufferIndex.putInt(size);    this.byteBufferIndex.putLong(tagsCode);     final long expectLogicOffset = cqOffset * CQ_STORE_UNIT_SIZE;     MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);    if (mappedFile != null) {        ...        return mappedFile.appendMessage(this.byteBufferIndex.array());    }}

3.2.3 写入 IndexFile

IndexFile 的文件逻辑构造如下图所示，相似于 JDK 的 HashMap 的数组加链表构造。次要由 Header、Slot Table、Index Linked List 三局部组成。

Header：IndexFile 的头部，占 40 个字节。次要蕴含以下字段：

beginTimestamp：该 IndexFile 文件中蕴含音讯的最小存储工夫。
endTimestamp：该 IndexFile 文件中蕴含音讯的最大存储工夫。
beginPhyoffset：该 IndexFile 文件中蕴含音讯的最小 CommitLog 文件偏移量。
endPhyoffset：该 IndexFile 文件中蕴含音讯的最大 CommitLog 文件偏移量。
hashSlotcount：该 IndexFile 文件中蕴含的 hashSlot 的总数。
indexCount：该 IndexFile 文件中已应用的 Index 条目个数。

Slot Table：默认蕴含 500w 个 Hash 槽，每个 Hash 槽存储的是雷同 hash 值的第一个 IndexItem 存储地位。

Index Linked List：默认最多蕴含 2000w 个 IndexItem。其组成如下所示：

Key Hash：音讯 key 的 hash，当依据 key 搜寻时比拟的是其 hash，在之后会比拟 key 自身。
CommitLog Offset：音讯的物理位移。
Timestamp：该音讯存储工夫与第一条音讯的工夫戳的差值。
Next Index Offset：产生 hash 抵触后保留的下一个 IndexItem 的地位。

Slot Table 中每个 hash 槽寄存的是 IndexItem 在 Index Linked List 的地位，如果 hash 抵触时，新的 IndexItem 插入链表头，它的 Next Index Offset 中寄存之前链表头 IndexItem 地位，同时笼罩 Slot Table 中的 hash 槽为最新 IndexItem 地位。代码如下：

//org.apache.rocketmq.store.index.IndexFile::putKeypublic boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {    int keyHash = indexKeyHashMethod(key);    int slotPos = keyHash % this.hashSlotNum;    int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;    ...    //从 Slot Table 获取以后最新消息地位    int slotValue = this.mappedByteBuffer.getInt(absSlotPos);    ...    int absIndexPos =        IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize            + this.indexHeader.getIndexCount() * indexSize;     this.mappedByteBuffer.putInt(absIndexPos, keyHash);    this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);    this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);    //寄存之前链表头 IndexItem 地位    this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);    //更新 Slot Table 中 hash 槽的值为最新消息地位    this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());     if (this.indexHeader.getIndexCount() <= 1) {        this.indexHeader.setBeginPhyOffset(phyOffset);        this.indexHeader.setBeginTimestamp(storeTimestamp);    }     if (invalidIndex == slotValue) {        this.indexHeader.incHashSlotCount();    }    this.indexHeader.incIndexCount();    this.indexHeader.setEndPhyOffset(phyOffset);    this.indexHeader.setEndTimestamp(storeTimestamp);     return true;    ...}

综上所述一个残缺的音讯写入流程包含：同步写入 Commitlog 文件缓存区，异步构建 ConsumeQueue、IndexFile 文件。

3.3 音讯刷盘

RocketMQ 音讯刷盘次要分为同步刷盘和异步刷盘。

(1) 同步刷盘：只有在音讯真正长久化至磁盘后 RocketMQ 的 Broker 端才会真正返回给 Producer 端一个胜利的 ACK 响应。同步刷盘对 MQ 音讯可靠性来说是一种不错的保障，然而性能上会有较大影响，个别金融业务应用该模式较多。

(2) 异步刷盘：可能充分利用 OS 的 Page Cache 的劣势，只有音讯写入 Page Cache 即可将胜利的 ACK 返回给 Producer 端。音讯刷盘采纳后盾异步线程提交的形式进行，升高了读写提早，进步了 MQ 的性能和吞吐量。异步刷盘蕴含开启堆外内存和未开启堆外内存两种形式。

在 CommitLog 中提交刷盘申请时，会依据以后 Broker 相干配置决定是同步刷盘还是异步刷盘。

//org.apache.rocketmq.store.CommitLog::submitFlushRequestpublic CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {    //同步刷盘    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {        final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;        if (messageExt.isWaitStoreMsgOK()) {            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);        }    }    //异步刷盘    else {        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {            flushCommitLogService.wakeup();        } else  {            //开启堆外内存的异步刷盘            commitLogService.wakeup();        }        return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);    }}

GroupCommitService、FlushRealTimeService、CommitRealTimeService 三者继承关系如图；

GroupCommitService：同步刷盘线程。如下图所示，音讯写入到 Page Cache 后通过 GroupCommitService 同步刷盘，音讯解决线程阻塞期待刷盘后果。

//org.apache.rocketmq.store.CommitLog.GroupCommitService::runpublic void run() {    ...    while (!this.isStopped()) {        this.waitForRunning(10);        this.doCommit();    }    ...} //org.apache.rocketmq.store.CommitLog.GroupCommitService::doCommitprivate void doCommit() {    ...    for (GroupCommitRequest req : this.requestsRead) {        boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();        for (int i = 0; i < 2 && !flushOK; i++) {            CommitLog.this.mappedFileQueue.flush(0);            flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();        }        //唤醒期待刷盘实现的音讯解决线程        req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);    }    ...} //org.apache.rocketmq.store.MappedFile::flushpublic int flush(final int flushLeastPages) {    if (this.isAbleToFlush(flushLeastPages)) {        ...        //应用到了 writeBuffer 或者 fileChannel 的 position 不为 0 时用 fileChannel 进行强制刷盘        if (writeBuffer != null || this.fileChannel.position() != 0) {            this.fileChannel.force(false);        } else {            //应用 MappedByteBuffer 进行强制刷盘            this.mappedByteBuffer.force();        }        ...    }}

FlushRealTimeService：未开启堆外内存的异步刷盘线程。如下图所示，音讯写入到 Page Cache 后，音讯解决线程立刻返回，通过 FlushRealTimeService 异步刷盘。

//org.apache.rocketmq.store.CommitLog.FlushRealTimeServicepublic void run() {    ...    //判断是否须要周期性进行刷盘    if (flushCommitLogTimed) {        //固定休眠 interval 工夫距离        Thread.sleep(interval);    } else {        // 如果被唤醒就刷盘，非周期性刷盘        this.waitForRunning(interval);    }    ...    // 这边和 GroupCommitService 用的是同一个强制刷盘办法    CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);    ...}

CommitRealTimeService：开启堆外内存的异步刷盘线程。如下图所示，音讯解决线程把音讯写入到堆外内存后立刻返回。后续先通过 CommitRealTimeService 把音讯由堆外内存异步提交至 Page Cache，再由 FlushRealTimeService 线程异步刷盘。

留神：在音讯异步提交至 Page Cache 后，业务就能够从 MappedByteBuffer 读取到该音讯。

音讯写入到堆外内存 writeBuffer 后，会通过 isAbleToCommit 办法判断是否积攒到至多提交页数（默认4页）。如果页数达到最小提交页数，则批量提交；否则还是驻留在堆外内存，这边有失落音讯危险。通过这种批量操作，读和写的 Page Cahe 会距离数页，升高了 Page Cahe 读写抵触的概率，实现了读写拆散。具体实现逻辑如下：

//org.apache.rocketmq.store.CommitLog.CommitRealTimeServiceclass CommitRealTimeService extends FlushCommitLogService {    @Override    public void run() {        while (!this.isStopped()) {            ...            int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();             ...            //把音讯 commit 到内存缓冲区，最终调用的是 MappedFile::commit0 办法，只有达到起码提交页数能力提交胜利，否则还在堆外内存中            boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);            if (!result) {                //唤醒 flushCommitLogService，进行强制刷盘                flushCommitLogService.wakeup();            }            ...            this.waitForRunning(interval);        }    }} //org.apache.rocketmq.store.MappedFile::commit0protected void commit0() {    int writePos = this.wrotePosition.get();    int lastCommittedPosition = this.committedPosition.get();         //音讯提交至 Page Cache，并未理论刷盘    if (writePos - lastCommittedPosition > 0) {        ByteBuffer byteBuffer = writeBuffer.slice();        byteBuffer.position(lastCommittedPosition);        byteBuffer.limit(writePos);        this.fileChannel.position(lastCommittedPosition);        this.fileChannel.write(byteBuffer);        this.committedPosition.set(writePos);    }}

上面总结一下三种刷盘机制的应用场景及优缺点。

四、音讯读取

音讯读取逻辑相比写入逻辑简略很多，上面着重剖析下依据 offset 查问音讯和依据 key 查问音讯是如何实现的。

4.1 依据 offset 查问

读取音讯的过程就是先从 ConsumeQueue 中找到音讯在 CommitLog 的物理偏移地址，而后再从 CommitLog 文件中读取音讯的实体内容。

//org.apache.rocketmq.store.DefaultMessageStore::getMessagepublic GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,    final int maxMsgNums,    final MessageFilter messageFilter) {    long nextBeginOffset = offset;     GetMessageResult getResult = new GetMessageResult();     final long maxOffsetPy = this.commitLog.getMaxOffset();    //找到对应的 ConsumeQueue    ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);    ...    //依据 offset 找到对应的 ConsumeQueue 的 MappedFile    SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset);    status = GetMessageStatus.NO_MATCHED_MESSAGE;    long maxPhyOffsetPulling = 0;     int i = 0;    //能返回的最大信息大小，不能大于 16M    final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE);    for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {        //CommitLog 物理地址        long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();        int sizePy = bufferConsumeQueue.getByteBuffer().getInt();        maxPhyOffsetPulling = offsetPy;        ...        //依据 offset 和 size 从 CommitLog 拿到具体的 Message        SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);        ...        //将 Message 放入后果集        getResult.addMessage(selectResult);        status = GetMessageStatus.FOUND;    }     //更新 offset    nextBeginOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);     long diff = maxOffsetPy - maxPhyOffsetPulling;    long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE        * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));    getResult.setSuggestPullingFromSlave(diff > memory);    ...    getResult.setStatus(status);    getResult.setNextBeginOffset(nextBeginOffset);    return getResult;}

4.2 依据 key 查问

读取音讯的过程就是用 topic 和 key 找到 IndexFile 索引文件中的一条记录，依据记录中的 CommitLog 的 offset 从 CommitLog 文件中读取音讯的实体内容。

//org.apache.rocketmq.store.DefaultMessageStore::queryMessagepublic QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end) {    QueryMessageResult queryMessageResult = new QueryMessageResult();    long lastQueryMsgTime = end;         for (int i = 0; i < 3; i++) {        //获取 IndexFile 索引文件中记录的音讯在 CommitLog 文件物理偏移地址        QueryOffsetResult queryOffsetResult = this.indexService.queryOffset(topic, key, maxNum, begin, lastQueryMsgTime);        ...        for (int m = 0; m < queryOffsetResult.getPhyOffsets().size(); m++) {            long offset = queryOffsetResult.getPhyOffsets().get(m);            ...            MessageExt msg = this.lookMessageByOffset(offset);            if (0 == m) {                lastQueryMsgTime = msg.getStoreTimestamp();            }            ...            //在 CommitLog 文件获取音讯内容            SelectMappedBufferResult result = this.commitLog.getData(offset, false);            ...            queryMessageResult.addMessage(result);            ...        }    }     return queryMessageResult;}

在 IndexFile 索引文件，查找 CommitLog 文件物理偏移地址实现如下：

//org.apache.rocketmq.store.index.IndexFile::selectPhyOffsetpublic void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum,final long begin, final long end, boolean lock) {    int keyHash = indexKeyHashMethod(key);    int slotPos = keyHash % this.hashSlotNum;    int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;    //获取雷同 hash 值 key 的第一个 IndexItme 存储地位，即链表的首节点    int slotValue = this.mappedByteBuffer.getInt(absSlotPos);         //遍历链表节点    for (int nextIndexToRead = slotValue; ; ) {        if (phyOffsets.size() >= maxNum) {            break;        }         int absIndexPos =            IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize                + nextIndexToRead * indexSize;         int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);        long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);         long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);        int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);         if (timeDiff < 0) {            break;        }         timeDiff *= 1000L;         long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;        boolean timeMatched = (timeRead >= begin) && (timeRead <= end);         //符合条件的后果退出 phyOffsets        if (keyHash == keyHashRead && timeMatched) {            phyOffsets.add(phyOffsetRead);        }         if (prevIndexRead <= invalidIndex            || prevIndexRead > this.indexHeader.getIndexCount()            || prevIndexRead == nextIndexToRead || timeRead < begin) {            break;        }                 //持续遍历链表        nextIndexToRead = prevIndexRead;    }    ...}

五、总结

本文从源码的角度介绍了 RocketMQ 存储系统的外围模块实现，包含存储架构、音讯写入和音讯读取。

RocketMQ 把所有 Topic 下的音讯都写入到 CommitLog 外面，实现了严格的程序写。通过文件预热避免 Page Cache 被替换到 swap 空间，缩小读写文件时缺页中断。应用 mmap 对 CommitLog 文件进行读写，将对文件的操作转化为间接对内存地址进行操作，从而极大地提高了文件的读写效率。

对于性能要求高、数据一致性要求不高的场景下，能够通过开启堆外内存，实现读写拆散，晋升磁盘的吞吐量。总之，存储模块的学习须要对操作系统原理有肯定理解。作者采纳的性能极致优化计划值得咱们好好学习。

六、参考文献

1.RocketMQ 官网文档

作者：vivo互联网服务器团队-Zhang Zhenglin