RocketMQ通过使用内存映射文件来提高I/O访问性能，无论是 CommitLog、Consume-Queue还是Index，单个文件都被设计为固定长 度，一个文件写满以后再创建新文件，文件名就为该文件第一条消息 对应的全局物理偏移量。

RocketMQ使用MappedFile、MappedFileQueue来封装存储文件。

1.1 MappedFileQueue映射文件队列

MappedFileQueue是MappedFile的管理容器，MappedFileQueue对 存储目录进行封装，例如CommitLog文件的存储路径为 ${ROCKET_HOME}/store/commitlog/，该目录下会存在多个内存映射文 件MappedFile。MappedFileQueue类图如图4-11所示。

图4-11 MappedFileQueue类图

下面介绍MappedFileQueue的核心属性。

1)String storePath：存储目录。

2)int mappedFileSize：单个文件的存储大小。 3)CopyOnWriteArrayList mappedFiles：MappedFile集合。

4)AllocateMappedFileService allocateMappedFileService： 创建MappedFile服务类。

5)long flushedWhere = 0：当前刷盘指针，表示该指针之前的 所有数据全部持久化到磁盘。

6)long committedWhere = 0：当前数据提交指针，内存中 ByteBuffer当前的写指针，该值大于、等于flushedWhere。

接下来重点分析根据不同维度查找MappedFile的方法，如代码清 单4-11所示。

代码清单4-11 MappedFileQueue#getMappedFileByTime

public MappedFile getMappedFileByTime(final long timestamp) {  Object[] mfs = this.copyMappedFiles(0);if (null == mfs)return null;for (int i = 0; i < mfs .length; i++) {  MappedFile mappedFile = (MappedFile) mfs[i];if (mappedFile .getLastModifiedTimestamp() >= timestamp) {return mappedFile;}}return (MappedFile) mfs[mfs.length - 1];}

根据消息存储时间戳查找MappdFile。从MappedFile列表中第一个 文件开始查找，找到第一个最后一次更新时间大于待查找时间戳的文 件，如果不存在，则返回最后一个MappedFile，如代码清单4-12所 示。

代码清单4-12 MappedFileQueue#findMappedFileByOffset

public MappedFile findMappedFileByOffset(final long offset, final booleanreturnFirstOnNotFound) {// 省略外层 try ...catchMappedFile mappedFile = this .getFirstMappedFile();if (mappedFile != null) {int index = (int) ((offset / this .mappedFileSize) -(mappedFile .getFileFromOffset() /this .mappedFileSize));if (index < 0 || index >= this .mappedFiles .size()) { // 省略警告日志}try {return this .mappedFiles .get(index);} catch (Exception e) {if (returnFirstOnNotFound) {return mappedFile;}LOG_ERROR.warn("findMappedFileByOffset failure. ", e); }}}

根据消息偏移量offset查找MappedFile，但是不能直接使用 offset%mappedFileSize。这是因为使用了内存映射，只要是存在于存 储目录下的文件，都需要对应创建内存映射文件，如果不定时将已消 费的消息从存储文件中删除，会造成极大的内存压力与资源浪费，所 以RocketMQ采取定时删除存储文件的策略。也就是说，在存储文件 中，第一个文件不一定是00000000000000000000，因为该文件在某一 时刻会被删除，所以根据offset定位MappedFile的算法为(int) ((offset/this.mappedFileSize)

(mappedFile.getFileFromOffset()/this.MappedFileSize))，如代码 清单4-13所示。

代码清单4-13 MappedFileQueue#getMinOffset

public long getMinOffset() {if (!this .mappedFiles .isEmpty()) { try {return this .mappedFiles .get(0) .getFileFromOffset(); } catch (IndexOutOfBoundsException e) {} catch (Exception e) {log .error("getMinOffset has exception .", e);

获取存储文件最小偏移量。从这里也可以看出，并不是直接返回 0，而是返回MappedFile的getFileFormOffset()方法，如代码清单4- 14所示。

代码清单4-14 MappedFileQueue#getMaxOffset

public long getMaxOffset() {MappedFile mappedFile = getLastMappedFile();if (mappedFile != null) {return mappedFile.getFileFromOffset() +mappedFile.getReadPosition();}return 0;}

获取存储文件的最大偏移量。返回最后一个MappedFile的 fileFromOffset，加上MappedFile当前的写指针，如代码清单4-15所 示。

代码清单4-15 MappedFileQueue#getMaxWrotePosition

public long getMaxWrotePosition() {MappedFile mappedFile = getLastMappedFile(); if (mappedFile !=null) {return mappedFile.getFileFromOffset() +mappedFile.getWrotePosition();}return 0;}

返回存储文件当前的写指针。返回最后一个文件的

fileFromOffset，加上当前写指针位置。

关于MappedFileQueue的相关业务方法，我们在具体使用到时再去

剖析。

1.2 MappedFile内存映射文件

MappedFile是RocketMQ内存映射文件的具体实现，如图4-12所 示。

图4-12 MappedFile类图

下面介绍MappedFile的核心属性。

1)int OS_PAGE_SIZE：操作系统每页大小，默认4KB。

2)AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY：当前JVM实例中 MappedFile的虚拟内存。

3)AtomicInteger TOTAL_MAPPED_FILES：当前JVM实例中 MappedFile对象个数。

4)AtomicInteger wrotePosition：当前文件的写指针，从0开始 (内存映射文件中的写指针)。

5)AtomicInteger committedPosition：当前文件的提交指针， 如果开启transientStore-PoolEnable，则数据会存储在 TransientStorePool中，然后提交到内存映射ByteBuffer中，再写入 磁盘。

6)AtomicInteger flushedPosition：将该指针之前的数据持久 化存储到磁盘中。

7)int fileSize：文件大小。

8)FileChannel fileChannel：文件通道。

9)ByteBuffer writeBuffer：堆外内存ByteBuffer，如果不为 空，数据首先将存储在该Buffer中，然后提交到MappedFile创建的 FileChannel中。transientStorePoolEnable为true时不为空。

10)TransientStorePool transientStorePool：堆外内存池，该 内存池中的内存会提供内存锁机制。transientStorePoolEnable为 true时启用。

11)String fileName：文件名称。

12)long fileFromOffset：该文件的初始偏移量。

13)File file：物理文件。

14)MappedByteBuffer mappedByteBuffer：物理文件对应的内存 映射Buffer。

15)volatile long storeTimestamp = 0：文件最后一次写入内 容的时间。

16)boolean firstCreateInQueue：是否是MappedFileQueue队列 中第一个文件。

1. MappedFile初始化

第一步：根据是否开启transientStorePoolEnable存在两种初始 化情况。transientStorePool-Enable为true表示内容先存储在堆外内 存，然后通过Commit线程将数据提交到FileChannel中，再通过Flush 线程将数据持久化到磁盘中，如代码清单4-16所示。

代码清单4-16 MappedFile#init(final String fileName, final int fileSize)

this.fileName = fileName;this.fileSize = fileSize;this.file = new File(fileName);this .fileFromOffset = Long .parseLong(this .file .getName()); ensureDirOK(this .file .getParent());this .fileChannel = new RandomAccessFile(this .file,"rw") .getChannel();this .mappedByteBuffer = this .fileChannel .map(MapMode .READ_WRITE, 0,fileSize);TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);TOTAL_MAPPED_FILES.incrementAndGet();

第二步：初始化fileFromOffset为文件名，也就是文件名代表该 文件的起始偏移量，通过RandomAccessFile创建读写文件通道，并将 文件内容使用NIO的内存映射Buffer将文件映射到内存中，如代码清单 4-17所示。

代码清单4-17 MappedFile#init

public void init(final String fileName, final int fileSize, final TransientStorePool transientStorePool) throwsIOException {init(fileName, fileSize);this.writeBuffer = transientStorePool.borrowBuffer();

如果transientStorePoolEnable为true，则初始化MappedFile的 writeBuffer，该buffer从transientStorePool中获取。

2. MappedFile提交

内存映射文件的提交动作由MappedFile的commit()方法实现，如 代码清单4-18所示。

代码清单4-18 MappedFile#commit

public int commit(final int commitLeastPages) {if (writeBuffer == null) {return this.wrotePosition.get();}if (this.isAbleToCommit(commitLeastPages)) {if (this.hold()) {commit0(commitLeastPages);this.release();} else {log .warn("in commit, hold failed, commit offset = "+this.committedPosition.get());}}if (writeBuffer != null && this.transientStorePool != null&&this.fileSize == this.committedPosition.get()) {this .transientStorePool .returnBuffer(writeBuffer);this .writeBuffer = null;}return this.committedPosition.get();}

执行提交操作，commitLeastPages为本次提交的最小页数，如果 待提交数据不满足commitLeastPages，则不执行本次提交操作，等待 下次提交。writeBuffer如果为空，直接返回wrotePosition指针，无 须执行commit操作，这表明commit操作的主体是writeBuffer，如代码 清单4-19所示。

代码清单4-19 MappedFile#isAbleToCommit

protected boolean isAbleToCommit(final int commitLeastPages) {int flush = this.committedPosition.get();int write = this.wrotePosition.get();if (this.isFull()) {return true;}if (commitLeastPages > 0) {return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) commitLeastPages;>=}return write > flush;}

判断是否执行commit操作。如果文件已满，返回true。如果 commitLeastPages大于0，则计算wrotePosition(当前writeBuffe的 写指针)与上一次提交的指针(committedPosition)的差值，将其除 以OS_PAGE_SIZE得到当前脏页的数量，如果大于commitLeastPages， 则返回true。如果commitLeastPages小于0，表示只要存在脏页就提 交，如代码清单4-20所示。

代码清单4-20 MappedFile#commit0

protected void commit0(final int commitLeastPages) { int writePos = this.wrotePosition.get(); int lastCommittedPosition = this.committedPosition.get(); if (writePos - this.committedPosition.get() > 0) {try {ByteBuffer byteBuffer = writeBuffer.slice(); byteBuffer.position(lastCommittedPosition); byteBuffer.limit(writePos);this.fileChannel.position(lastCommittedPosition); this.fileChannel.write(byteBuffer);this.committedPosition.set(writePos); }catch (Throwable e) {log.error("Error occurred when commit data to FileChannel.", e);}}}

下面介绍具体的MappedFile提交实现过程。首先创建writeBuffer 的共享缓存区，然后将新创建的position回退到上一次提交的位置 (committedPosition)，设置limit为wrotePosition(当前最大有效 数据指针)，接着把committedPosition到wrotePosition的数据复制

(写入)到FileChannel中，最后更新committedPosition指针为 wrotePosition。commit的作用是将MappedFile# writeBuffer中的数 据提交到文件通道FileChannel中。

ByteBuffer使用技巧：调用slice()方法创建一个共享缓存区，与 原先的ByteBuffer共享内存并维护一套独立的指针(position、 mark、limit)。

3. MappedFile刷盘

刷盘指的是将内存中的数据写入磁盘，永久存储在磁盘中，由 MappedFile的flush()方法实现，如代码清单4-21所示。

代码清单4-21 MappedFile#flush

public int flush(final int flushLeastPages) { if(this.isAbleToFlush(flushLeastPages)) {if (this.hold()) {int value = getReadPosition();try {if (writeBuffer != null ||this.fileChannel.position() != 0) {this.fileChannel.force(false);} else {this.mappedByteBuffer.force();}} catch (Throwable e) {log.error("Error occurred when force data todisk.", e);}this.flushedPosition.set(value);this.release();} else {this.flushedPosition.set(getReadPosition()); }}return this.getFlushedPosition();}

直接调用mappedByteBuffer或fileChannel的force()方法将数据 写入磁盘，将内存中的数据持久化到磁盘中，那么flushedPosition应 该等于MappedByteBuffer中的写指针。如果writeBuffer不为空，则 flushedPosition应等于上一次commit指针。因为上一次提交的数据就 是进入MappedByteBuffer中的数据。如果writeBuffer为空，表示数据

是直接进入MappedByteBuffer的，wrotePosition代表的是 MappedByteBuffer中的指针，故设置flushedPosition为 wrotePosition。

4. 获取MappedFile最大读指针

RocketMQ文件的一个组织方式是内存映射，预先申请一块连续且 固定大小的内存，需要一套指针标识当前最大有效数据的位置，获取 最大有效数据偏移量的方法由MappedFile的getReadPosition()方法实 现，如代码清单4-22所示。

代码清单4-22 MappedFile#getReadPosition

public int getReadPosition() {return this.writeBuffer == null ? this.wrotePosition.get() : this.committedPosition.get();}

获取当前文件最大的可读指针，如代码清单4-23所示。如果 writeBuffer为空，则直接返回当前的写指针。如果writeBuffer不为 空，则返回上一次提交的指针。在MappedFile设计中，只有提交了的 数据(写入MappedByteBuffer或FileChannel中的数据)才是安全的数 据。

代码清单4-23 MappedFile#selectMappedBuffer

public SelectMappedBufferResult selectMappedBuffer(int pos) {int readPosition = getReadPosition();if (pos < readPosition && pos >= 0) { if(this.hold()) {ByteBuffer byteBuffer = this.mappedByteBuffer.slice(); byteBuffer.position(pos);int size = readPosition - pos;ByteBuffer byteBufferNew = byteBuffer.slice(); byteBufferNew.limit(size);return new SelectMappedBufferResult(this.fileFromOffset+ pos,byteBufferNew, size, this);}}return null;}

首先查找pos到当前最大可读指针之间的数据，因为在整个写入期 间都未曾改变MappedByteBuffer的指针，所以

mappedByteBuffer.slice()方法返回的共享缓存区空间为整个 MappedFile。然后通过设置byteBuffer的position为待查找的值，读 取字节为当前可读字节长度，最终返回的ByteBuffer的limit(可读最 大长度)为size。整个共享缓存区的容量为

MappedFile#fileSizepos，故在操作SelectMappedBufferResult时不 能对包含在里面的ByteBuffer调用flip()方法。

注意

操作ByteBuffer时如果使用了slice()方法，对其ByteBuffer进行读取 时一般手动指定position和limit指针，而不是调用flip()方法切换读 写状态。

5. MappedFile销毁

MappedFile文件销毁的实现方法为public boolean destroy(final long intervalForcibly)，intervalForcibly表示拒 绝被销毁的最大存活时间，如代码清单4-24所示。

代码清单4-24 MappedFile#shutdown

public void shutdown(final long intervalForcibly) { if(this.available) {this.available = false;this.firstShutdownTimestamp = System.currentTimeMillis(); this.release();} else if (this.getRefCount() > 0) {if ((System.currentTimeMillis() -this.firstShutdownTimestamp) >=intervalForcibly) {this.refCount.set(-1000 - this.getRefCount()); this.release();}}}

第一步：关闭MappedFile。初次调用时this.available为true， 设置available为false，并设置初次关闭的时间戳 (firstShutdownTimestamp)为当前时间戳。调用release()方法尝试

释放资源，release只有在引用次数小于1的情况下才会释放资源。如 果引用次数大于0，对比当前时间与firstShutdownTimestamp，如果已 经超过了其最大拒绝存活期，则每执行一次引用操作，引用数减少 1000，直到引用数小于0时通过执行realse()方法释放资源，如代码清 单4-25所示。

代码清单4-25 MappedFile#isCleanupOver

public boolean isCleanupOver() {return this.refCount.get() <= 0 && this.cleanupOver; }

第二步：判断是否清理完成，判断标准是引用次数小于、等于0并 且cleanupOver为true，cleanupOver为true的触发条件是release成功 将MappedByteBuffer资源释放了，如代码清单4-26所示。稍后详细分 析release()方法。

代码清单4-26 MappedFile#destroy

this.fileChannel.close();log.info("close file channel " + this.fileName + " OK");long beginTime = System.currentTimeMillis();boolean result = this.file.delete();

第三步：关闭文件通道，删除物理文件。

在整个MappedFile销毁的过程中，首先需要释放资源，释放资源 的前提条件是该MappedFile的引用小于、等于0。接下来重点看一下 release()方法的实现原理，如代码清单4-27所示。

代码清单4-27 ReferenceResource#release

public void release() {long value = this.refCount.decrementAndGet();if (value > 0)return;synchronized (this) {this.cleanupOver = this.cleanup(value);}}

将引用次数减1，如果引用数小于、等于0，则执行cleanup()方 法，下面重点分析cleanup()方法的实现，如代码清单4-28所示。

代码清单4-28 MappedFile#cleanup

public boolean cleanup(final long currentRef) {   if(this .isAvailable()) {return false;}if (this.isCleanupOver()) {return true;}clean(this .mappedByteBuffer);TOTAL_MAPPED_VIRTUAL_MEMORY .addAndGet(this .fileSize * (-1)); TOTAL_MAPPED_FILES .decrementAndGet();log .info("unmap file[REF:" + currentRef + "] " + this .fileName+ " OK");return true;}

如果available为true，表示MappedFile当前可用，无须清理，返 回false，如果资源已经被清除，返回true。如果是堆外内存，调用堆 外内存的cleanup()方法进行清除，维护MappedFile类变量 TOTAL_MAPPED_VIRTUAL_MEMORY、TOTAL_MAPPED_FILES并返回true，表 示cleanupOver为true。

1.3 TransientStorePool

TransientStorePool即短暂的存储池。RocketMQ单独创建了一个 DirectByteBuffer内存缓存池，用来临时存储数据，数据先写入该内 存映射中，然后由Commit线程定时将数据从该内存复制到与目标物理 文件对应的内存映射中。RokcetMQ引入该机制是为了提供一种内存锁 定，将当前堆外内存一直锁定在内存中，避免被进程将内存交换到磁 盘中。

TransientStorePool类图如图4-13所示。

图4-13 TransientStorePool类图

下面介绍TransientStorePool的核心属性，如代码清单4-29所 示。

1)int poolSize：avaliableBuffers个数，可在broker配置文件 中通过transient StorePoolSize进行设置，默认为5。

2)int fileSize：每个ByteBuffer的大小，默认为 mapedFileSizeCommitLog，表明TransientStorePool为CommitLog文件 服务。

3)Deque availableBuffers：ByteBuffer容器，双端队列。 代码清单4-29 TransientStorePool#init

public void init() {for (int i = 0; i < poolSize; i++) {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);}}

创建数量为poolSize的堆外内存，利用com.sun.jna.Library类库 锁定该批内存，避免被置换到交换区，以便提高存储性能。

2 RocketMQ存储文件

RocketMQ存储路径为${ROCKET_HOME}/store，主要存储文件如图 4-14所示。下面介绍RocketMQ主要的存储文件夹。

图4-14 RocketMQ存储 目 录 1)commitlog：消息存储目录。 2)config：运行期间的一些配置信息，主要包括下列信息。

consumerFilter.json：主题消息过滤信息。

consumerOffset.json：集群消费模式下的消息消费进度。

delayOffset.json：延时消息队列拉取进度。

subscriptionGroup.json：消息消费组的配置信息。

topics.json：topic配置属性。

3)consumequeue：消息消费队列存储目录。

4)index：消息索引文件存储目录。

5)abort：如果存在abort文件，说明Broker非正常关闭，该文件 默认在启动Broker时创建，在正常退出之前删除。

6)checkpoint：检测点文件，存储CommitLog文件最后一次刷盘 时间戳、ConsumeQueue最后一次刷盘时间、index文件最后一次刷盘时 间戳。

2.1 CommitLog文件

CommitLog目录的结构在1节已经详细介绍过了，该目录下的文 件主要用于存储消息，其特点是每一条消息长度不相同。CommitLog文 件存储格式如图4-15所示，每条消息的前面4个字节存储该条消息的总 长度。

图4-15 CommitLog文件存储格式

CommitLog文件的存储目录默认为

${ROCKET_HOME}/store/commitlog，可以通过在broker配置文件中设 置storePathRootDir属性改变默认路径，如代码清单4-30所示。 CommitLog文件默认大小为1GB，可通过在broker配置文件中设置 mapedFileSizeCommitLog属性改变默认大小。本节将基于上述存储结 构，重点分析消息的查找实现。

代码清单4-30 Commitlog#getMinOffset

public long getMinOffset() {MappedFile mappedFile =this.mappedFileQueue.getFirstMappedFile();if (mappedFile != null) {if (mappedFile.isAvailable()) {return mappedFile.getFileFromOffset();} else {returnthis.rollNextFile(mappedFile.getFileFromOffset());}}return -1;}

获取当前CommitLog目录的最小偏移量，首先获取目录下的第一个 文件，如果该文件可用，则返回该文件的起始偏移量，否则返回下一 个文件的起始偏移量，如代码清单4-31所示。

代码清单4-31 CommitLog#rollNextFile

public long rollNextFile(final long offset) {int mappedFileSize =this.defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog();return offset + mappedFileSize - offset % mappedFileSize; }

根据offset返回下一个文件的起始偏移量。获取一个文件的大 小，减去offset % mapped-FileSize，回到下一文件的起始偏移量， 如代码清单4-32所示。

代码清单4-32 CommitLog#getMessage

public SelectMappedBufferResult getMessage(final long offset, final int size) { intmappedFileSize =this.defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog();MappedFile mappedFile =this.mappedFileQueue.findMappedFileByOffset(offset,offset == 0);if (mappedFile != null) {

根据偏移量与消息长度查找消息。首先根据偏移找到文件所在的 物理偏移量，然后用offset与文件长度取余，得到在文件内的偏移 量，从该偏移量读取size长度的内容并返回。如果只根据消息偏移量 查找消息，则首先找到文件内的偏移量，然后尝试读取4字节，获取消 息的实际长度，最后读取指定字节。

2.2 ConsumeQueue文件

RocketMQ基于主题订阅模式实现消息消费，消费者关心的是一个 主题下的所有消息，但同一主题的消息是不连续地存储在CommitLog文 件中的。如果消息消费者直接从消息存储文件中遍历查找订阅主题下 的消息，效率将极其低下。RocketMQ为了适应消息消费的检索需求， 设计了ConsumeQueue文件，该文件可以看作CommitLog关于消息消费的 “索引”文件，ConsumeQueue的第一级目录为消息主题，第二级目录 为主题的消息队列，如图4-16所示。

图4-16 ConsumeQueue文件结构

为了加速ConsumeQueue消息条目的检索速度并节省磁盘空间，每 一个ConsumeQueue条目不会存储消息的全量信息，存储格式如图4-17 所示。

图4-17 ConsumeQueue文件存储格式

单个ConsumeQueue文件中默认包含30万个条目，单个文件的长度 为3×106 ×20字节，单个ConsumeQueue文件可以看作一个ConsumeQueue 条目的数组，其下标为ConsumeQueue的逻辑偏移量，消息消费进度存 储的偏移量即逻辑偏移量。ConsumeQueue即为CommitLog文件的索引文 件，其构建机制是当消息到达CommitLog文件后，由专门的线程产生消 息转发任务，从而构建ConsumeQueue文件与下文提到的Index文件，如 代码清单4-33所示。本节只分析如何根据消息逻辑偏移量、时间戳查 找消息，4.6节将重点讨论消息消费队列的构建、恢复等内容。

代码清单4-33 ConsumeQueue#getIndexBuffer

public SelectMappedBufferResult getIndexBuffer(final long startIndex) {int mappedFileSize = this.mappedFileSize;long offset = startIndex *CQ_STORE_UNIT_SIZE;if (offset >= this.getMinLogicOffset()) {MappedFile mappedFile =this.mappedFileQueue.findMappedFileByOffset(offset);if (mappedFile != null) {SelectMappedBufferResult result =mappedFile.selectMappedBuffer((int)(offset % mappedFileSize));return result;}}return null;}

根据startIndex获取消息消费队列条目。通过startIndex×20得 到在ConsumeQueue文件的物理偏移量，如果该偏移量小于 minLogicOffset，则返回null，说明该消息已被删除，如果大于 minLogicOffset，则根据偏移量定位到具体的物理文件。通过将该偏 移量与物理文件的大小取模获取在该文件的偏移量，从偏移量开始连 续读取20个字节即可。

ConsumeQueue文件提供了根据消息存储时间来查找具体实现的算 法getOffsetInQueue- ByTime(final long timestamp)，其具体实现 如下。

第一步：根据时间戳定位到物理文件，就是从第一个文件开始， 找到第一个文件更新时间大于该时间戳的文件，如代码清单4-34所

示。

代码清单4-34 ConsumeQueue#getOffsetInQueueByTime

int low = minLogicOffset > mappedFile.getFileFromOffset() ? (int) (minLogicOffset- mappedFile.getFileFromOffset()) : 0;int high = 0;int midOffset = -1, targetOffset = -1, leftOffset = -1, rightOffset = -1;long leftIndexValue = -1L, rightIndexValue = -1L;long minPhysicOffset = this.defaultMessageStore.getMinPhyOffset(); SelectMappedBufferResult sbr = mappedFile.selectMappedBuffer(0);if (null != sbr) {ByteBuffer byteBuffer = sbr.getByteBuffer(); high = byteBuffer.limit() - CQ_STORE_UNIT_SIZE;

第二步：采用二分查找来加速检索。首先计算最低查找偏移量， 取消息队列最小偏移量与该文件注销偏移量的差为最小偏移量low。获 取当前存储文件中有效的最小消息物理偏移量minPhysicOffset，如果 查找到的消息偏移量小于该物理偏移量，则结束该查找过程，如代码 清单4-35所示。

代码清单4-35 ConsumeQueue#getOffsetInQueueByTime

while (high >= low) {midOffset = (low + high) / (2 * CQ_STORE_UNIT_SIZE) * CQ_STORE_UNIT_SIZE;byteBuffer.position(midOffset);long phyOffset = byteBuffer.getLong();int size = byteBuffer.getInt();if (phyOffset < minPhysicOffset) {low = midOffset + CQ_STORE_UNIT_SIZE;leftOffset = midOffset;continue;}long storeTime =this.defaultMessageStore.getCommitLog().pickupStoreTimestamp(phyOffset, size);if (storeTime < 0) {return 0;} else if (storeTime == timestamp) {targetOffset = midOffset; break;} else if (storeTime > timestamp) {high = midOffset - CQ_STORE_UNIT_SIZE;rightOffset = midOffset;

rightIndexValue = storeTime;} else {low = midOffset + CQ_STORE_UNIT_SIZE;leftOffset = midOffset;leftIndexValue = storeTime;}}

二分查找的常规退出循环为low>high，首先查找中间的偏移量 midOffset，将ConsumeQueue文件对应的ByteBuffer定位到 midOffset，然后读取4个字节，获取该消息的物理偏移量，如代码清 单4-36所示。

1)如果得到的物理偏移量小于当前的最小物理偏移量，说明待查 找消息的物理偏移量肯定大于midOffset，则将low设置为midOffset， 继续折半查找。

2)如果得到的物理偏移量大于最小物理偏移量，说明该消息是有 效消息，则根据消息偏移量和消息长度获取消息的存储时间戳。

3)如果存储时间小于0，则为无效消息，直接返回0。

4)如果存储时间戳等于待查找时间戳，说明查找到了匹配消息， 则设置targetOffset并跳出循环。

5)如果存储时间戳大于待查找时间戳，说明待查找消息的物理偏 移量小于midOffset，则设置high为midOffset，并设置 rightIndexValue等于midOffset。

6)如果存储时间戳小于待查找时间戳，说明待查找消息的物理偏 移量大于midOffset，则设置low为midOffset，并设置leftIndexValue 等于midOffset。

代码清单4-36 ConsumeQueue#getOffsetInQueueByTime

if (targetOffset != -1) {offset = targetOffset;} else {if (leftIndexValue == -1) {offset = rightOffset;} else if (rightIndexValue == -1) {offset = leftOffset;  } else {offset = Math.abs(timestamp - leftIndexValue) > Math.abs(timestamp- rightIndexValue) ? rightOffset : leftOffset;}}return (mappedFile.getFileFromOffset() + offset) /CQ_STORE_UNIT_SIZE;

第三步：如果targetOffset不等于-1，表示找到了存储时间戳等 于待查找时间戳的消息。如果leftIndexValue等于-1，表示返回当前 时间戳大于待查找消息的时间戳，并且最接近待查找消息的偏移量。 如果rightIndexValue等于-1，表示返回的时间戳比待查找消息的时间 戳小，并且最接近待查找消息的偏移量，如代码清单4-37所示。

代码清单4-37 ConsumeQueue#rollNextFile

public long rollNextFile(final long index) { int mappedFileSize = this.mappedFileSize;                              int totalUnitsInFile = mappedFileSize / CQ_STORE_UNIT_SIZE;                                             return index + totalUnitsInFile - index % totalUnitsInFile;}

根据当前偏移量获取下一个文件的起始偏移量。首先获取文件包 含多少个消息消费队列条目，减去index%totalUnitsInFile的目的是 选中下一个文件的起始偏移量。

2.3 Index文件

ConsumeQueue是RocketMQ专门为消息订阅构建的索引文件，目的 是提高根据主题与消息队列检索消息的速度。另外，RocketMQ引入哈 希索引机制为消息建立索引，HashMap的设计包含两个基本点：哈希槽 与哈希冲突的链表结构。RocketMQ索引文件Index存储格式如图4-18所 示。

图4-18 Index文件存储格式

从图4-20可以看出，Index包含Index文件头、哈希槽、Index条目 (数据)。Index文件头包含40字节，记录该Index的统计信息，其结 构如下。

1)beginTimestamp：Index文件中消息的最小存储时间。 2)endTimestamp：Index文件中消息的最大存储时间。

3)beginPhyoffset：Index文件中消息的最小物理偏移量 (CommitLog文件偏移量)。

4)endPhyoffset：Index文件中消息的最大物理偏移量 (CommitLog文件偏移量)。

5)hashslotCount：hashslot个数，并不是哈希槽使用的个数， 在这里意义不大。

6)indexCount：Index条目列表当前已使用的个数，Index条目在 Index条目列表中按顺序存储。

一个Index默认包含500万个哈希槽。哈希槽存储的是落在该哈希 槽的哈希码最新的Index索引。默认一个Index文件包含2000万个条 目，每个Index条目结构如下。

1)hashcode：key的哈希码。

2)phyoffset：消息对应的物理偏移量。

3)timedif：该消息存储时间与第一条消息的时间戳的差值，若 小于0，则该消息无效。

4)pre index no：该条目的前一条记录的Index索引，当出现哈 希冲突时，构建链表结构。

接下来重点分析如何将Map<String/*消息索引key*/,long phyOffset/*消息物理偏移量*/>存入Index文件，以及如何根据消息索 引key快速查找消息。

RocketMQ将消息索引键与消息偏移量的映射关系写入Index的实现 方法为public boolean putKey(final String key, final long phyOffset, final long storeTimestamp)，参数含义分别为消息索 引、消息物理偏移量、消息存储时间，如代码清单4-38所示。

代码清单4-38 IndexFile#putKey

if (this.indexHeader.getIndexCount() < this.indexNum) { int keyHash = indexKeyHashMethod(key);int slotPos = keyHash % this .hashSlotNum;int absSlotPos = IndexHeader .INDEX_HEADER_SIZE + slotPos *hashSlotSize;}

第一步：当前已使用条目大于、等于允许最大条目数时，返回 fasle，表示当前Index文件已写满。如果当前index文件未写满，则根 据key算出哈希码。根据keyHash对哈希槽数量取余定位到哈希码对应 的哈希槽下标，哈希码对应的哈希槽的物理地址为IndexHeader(40字 节)加上下标乘以每个哈希槽的大小(4字节)，如代码清单4-39所 示。

代码清单4-39 IndexFile#putKey

int slotValue = this.mappedByteBuffer.getInt(absSlotPos); if (slotValue <= invalidIndex || slotValue >this.indexHeader.getIndexCount()){slotValue = invalidIndex;}

第二步：读取哈希槽中存储的数据，如果哈希槽存储的数据小于0 或大于当前Index文件中的索引条目，则将slotValue设置为0，如代码 清单4-40所示。

代码清单4-40 IndexFile#putKey

long timeDiff = storeTimestamp -this.indexHeader.getBeginTimestamp();timeDiff = timeDiff / 1000;if (this.indexHeader.getBeginTimestamp() <= 0) {timeDiff = 0;} else if (timeDiff > Integer .MAX_VALUE) {timeDiff = Integer .MAX_VALUE;} else if (timeDiff < 0) {timeDiff = 0;}

第三步：计算待存储消息的时间戳与第一条消息时间戳的差值， 并转换成秒，如代码清单4-41所示。

代码清单4-41 IndexFile#putKey

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); this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue); this.mappedByteBuffer.putInt(absSlotPos,this .indexHeader .getIndexCount());

第四步：将条目信息存储在Index文件中。

1)计算新添加条目的起始物理偏移量：头部字节长度+哈希槽数 量×单个哈希槽大小(4个字节)+当前Index条目个数×单个Index条 目大小(20个字节)。

2)依次将哈希码、消息物理偏移量、消息存储时间戳与Index文 件时间戳、当前哈希槽的值存入MappedByteBuffer。

3)将当前Index文件中包含的条目数量存入哈希槽中，覆盖原先 哈希槽的值。

以上是哈希冲突链式解决方案的关键实现，哈希槽中存储的是该 哈希码对应的最新Index条目的下标，新的Index条目最后4个字节存储 该哈希码上一个条目的Index下标。如果哈希槽中存储的值为0或大于 当前Index文件最大条目数或小于-1，表示该哈希槽当前并没有与之对 应的Index条目。值得注意的是，Index文件条目中存储的不是消息索 引key，而是消息属性key的哈希，在根据key查找时需要根据消息物理 偏移量找到消息，进而验证消息key的值。之所以只存储哈希，而不存 储具体的key，是为了将Index条目设计为定长结构，才能方便地检索 与定位条目，如代码清单4-42所示。

代码清单4-42 IndexFile#putKey

if (this .indexHeader .getIndexCount() <= 1) {this .indexHeader .setBeginPhyOffset(phyOffset); this .indexHeader .setBeginTimestamp(storeTimestamp);}this.indexHeader.incHashSlotCount();this.indexHeader.incIndexCount();this.indexHeader.setEndPhyOffset(phyOffset);this.indexHeader.setEndTimestamp(storeTimestamp);

第五步：更新文件索引头信息。如果当前文件只包含一个条目， 则更新beginPhyOffset、beginTimestamp、endPyhOffset、 endTimestamp以及当前文件使用索引条目等信息，如代码清单4-43所 示。

RocketMQ根据索引key查找消息的实现方法为 selectPhyOffset(List<Long> phy Offsets, String key, int maxNum,long begin, long end)，其参数说明如下。

1)List<Long> phyOffsets：查找到的消息物理偏移量。 2)String key：索引key。

3)int maxNum：本次查找最大消息条数。

4)long begin：开始时间戳。

5)long end：结束时间戳。

代码清单4-43 IndexFile#selectPhyOffset

int keyHash = indexKeyHashMethod(key);int slotPos = keyHash % this.hashSlotNum;int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos *hashSlotSize;

第一步：根据key算出key的哈希码，keyHash对哈希槽数量取余， 定位到哈希码对应的哈希槽下标，哈希槽的物理地址为 IndexHeader(40字节)加上下标乘以每个哈希槽的大小(4字节)， 如代码清单4-44所示。

代码清单4-44 IndexFile#selectPhyOffset

int slotValue = this.mappedByteBuffer.getInt(absSlotPos); if (slotValue <= invalidIndex || slotValue >this.indexHeader.getIndexCount()|| this.indexHeader.getIndexCount() <= 1) { // 返回; }

第二步：如果对应的哈希槽中存储的数据小于1或大于当前索引条 目个数，表示该哈希码没有对应的条目，直接返回，如代码清单4-45 所示。

代码清单4-45 IndexFile#selectPhyOffset

for (int nextIndexToRead = slotValue; ; ){// 省略部分代码}

第三步：因为会存在哈希冲突，所以根据slotValue定位该哈希槽 最新的一个Item条目，将存储的物理偏移量加入phyOffsets，然后继 续验证Item条目中存储的上一个Index下标，如果大于、等于1并且小 于当前文件的最大条目数，则继续查找，否则结束查找，如代码清单 4-46所示。

代码清单4-46 IndexFile#selectPhyOffset

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);

第四步：根据Index下标定位到条目的起始物理偏移量，然后依次 读取哈希码、物理偏移量、时间戳、上一个条目的Index下标，如代码 清单4-47所示。

代码清单4-47 查找消息偏移量

if (timeDiff < 0) {break;}timeDiff *= 1000L;long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff; boolean timeMatched = (timeRead >= begin) && (timeRead <= end);if (keyHash == keyHashRead && timeMatched) {  phyOffsets.add(phyOffsetRead);}if (prevIndexRead <= invalidIndex|| prevIndexRead > this.indexHeader.getIndexCount() || prevIndexRead == nextIndexToRead || timeRead < begin) {break;}nextIndexToRead = prevIndexRead;

第五步：如果存储的时间戳小于0，则直接结束查找。如果哈希匹 配并且消息存储时间介于待查找时间start、end之间，则将消息物理 偏移量加入phyOffsets，并验证条目的前一个Index索引，如果索引大 于、等于1并且小于Index条目数，则继续查找，否则结束查找。

2.4 checkpoint文件

checkpoint(检查点)文件的作用是记录ComitLog、 ConsumeQueue、Index文件的刷盘时间点，文件固定长度为4KB，其中 只用该文件的前面24字节，其存储格式如图4-19所示。

图4-19 checkpoint文件存储格式 1)physicMsgTimestamp：CommitLog文件刷盘时间点。 2)logicsMsgTimestamp ：ConsumeQueue文件刷盘时间点。 3)indexMsgTimestamp：Index文件刷盘时间点。