该文所涉及的 RocketMQ 源码版本为 4.9.3。

RocketMQ MappedFile 内存映射文件详解

1、MappedFile 初始化

  1. private void init(final String fileName, final int fileSize) throws IOException {
  2.     this.fileName = fileName;
  3.     this.fileSize = fileSize;
  4.     this.file = new File(fileName);
  5.     this.fileFromOffset = Long.parseLong(this.file.getName());
  6.     boolean ok = false;
  8.     ensureDirOK(this.file.getParent());
  10.     try {
  11.         this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
  12.         this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
  13.         TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
  14.         TOTAL_MAPPED_FILES.incrementAndGet();
  15.         ok = true;
  16.     } catch (FileNotFoundException e) {
  17.         log.error("Failed to create file " + this.fileName, e);
  18.         throw e;
  19.     } catch (IOException e) {
  20.         log.error("Failed to map file " + this.fileName, e);
  21.         throw e;
  22.     } finally {
  23.         if (!ok && this.fileChannel != null) {
  24.             this.fileChannel.close();
  25.         }
  26.     }
  27. }

初始化fileFromOffset,因为 commitLog 文件夹下的文件都是以偏移量为命名的，所以转成了 long 类型

确认文件目录是否存在，不存在则创建

  1. public static void ensureDirOK(final String dirName) {
  2.     if (dirName != null) {
  3.         if (dirName.contains(MessageStoreConfig.MULTI_PATH_SPLITTER)) {
  4.             String[] dirs = dirName.trim().split(MessageStoreConfig.MULTI_PATH_SPLITTER);
  5.             for (String dir : dirs) {
  6.                 createDirIfNotExist(dir);
  7.             }
  8.         } else {
  9.             createDirIfNotExist(dirName);
  10.         }
  11.     }
  12. }

通过RandomAccessFile设置 fileChannel

this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();

使用 NIO 内存映射将文件映射到内存中

this.mappedByteBuffer = this.fileChannel.map(MapMode.*READ_WRITE*, 0, fileSize);

2、MappedFile 提交

  1. public int commit(final int commitLeastPages) {
  2.     if (writeBuffer == null) {
  3.         //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
  4.         return this.wrotePosition.get();
  5.     }
  6.     if (this.isAbleToCommit(commitLeastPages)) {
  7.         if (this.hold()) {
  8.             commit0();
  9.             this.release();
  10.         } else {
  11.             log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
  12.         }
  13.     }
  15.     // All dirty data has been committed to FileChannel.
  16.     if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
  17.         this.transientStorePool.returnBuffer(writeBuffer);
  18.         this.writeBuffer = null;
  19.     }
  21.     return this.committedPosition.get();
  22. }

如果 wroteBuffer 为空，直接返回 wrotePosition

  1. if (writeBuffer == null) {
  2.     //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
  3.     return this.wrotePosition.get();
  4. }

判断是否执行 commit 操作：

如果文件已满，返回 true

  1. if (this.isFull()) {
  2.     return true;
  3. }
  1. public boolean isFull() {
  2.     return this.fileSize == this.wrotePosition.get();
  3. }

commitLeastPages 为本次提交的最小页数，如果 commitLeastPages 大于 0，计算当前写指针（wrotePosition）与上一次提交的指针committedPosition的差值 除以页OS_PAGE_SIZE的大小得到脏页数量，如果大于 commitLeastPages，就可以提交。如果 commitLeastPages 小于 0，则存在脏页就提交

  1. if (commitLeastPages > 0) {
  2.     return ((write /OS_PAGE_SIZE) - (flush /OS_PAGE_SIZE)) >= commitLeastPages;
  3. }
  5. return write > flush;

MapperFile 具体的提交过程，首先创建 writeBuffer的共享缓存区，设置 position 为上一次提交的位置committedPosition ，设置 limit 为wrotePosition当前写指针，接着将 committedPosition 到 wrotePosition 的数据写入到 FileChannel 中，最后更新 committedPosition 指针为 wrotePosition

  1. protected void commit0() {
  2.     int writePos = this.wrotePosition.get();
  3.     int lastCommittedPosition = this.committedPosition.get();
  5.     if (writePos - lastCommittedPosition > 0) {
  6.         try {
  7.             ByteBuffer byteBuffer = writeBuffer.slice();
  8.             byteBuffer.position(lastCommittedPosition);
  9.             byteBuffer.limit(writePos);
  10.             this.fileChannel.position(lastCommittedPosition);
  11.             this.fileChannel.write(byteBuffer);
  12.             this.committedPosition.set(writePos);
  13.         } catch (Throwable e) {
  14.             log.error("Error occurred when commit data to FileChannel.", e);
  15.         }
  16.     }
  17. }

3、MappedFile 刷盘

判断是否要进行刷盘

文件是否已满

  1. if (this.isFull()) {
  2.     return true;
  3. }
  1. public boolean isFull() {
  2.     return this.fileSize == this.wrotePosition.get();
  3. }

如果flushLeastPages大于 0，判断写数据指针位置-上次刷盘的指针位置， 然后除以OS_PAGE_SIZE 是否大于等于flushLeastPages

如果 flushLeastPages 小于等于 0，判断是否有要刷盘的数据

  1. if (flushLeastPages > 0) {
  2.     return ((write /OS_PAGE_SIZE) - (flush /OS_PAGE_SIZE)) >= flushLeastPages;
  3. }
  5. return write > flush;

获取最大读指针

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

将数据刷出到磁盘

如果writeBuffer不为空或者通道的 position 不等于 0，通过 fileChannel 将数据刷新到磁盘

否则通过 MappedByteBuffer 将数据刷新到磁盘

4、MappedFile 销毁

  1. public boolean destroy(final long intervalForcibly) {
  2.     this.shutdown(intervalForcibly);
  4.     if (this.isCleanupOver()) {
  5.         try {
  6.             this.fileChannel.close();
  7.             log.info("close file channel " + this.fileName + " OK");
  9.             long beginTime = System.currentTimeMillis();
  10.             boolean result = this.file.delete();
  11.             log.info("delete file[REF:" + this.getRefCount() + "] " + this.fileName
  12.                 + (result ? " OK, " : " Failed, ") + "W:" + this.getWrotePosition() + " M:"
  13.                 + this.getFlushedPosition() + ", "
  14.                 + UtilAll.computeElapsedTimeMilliseconds(beginTime));
  15.         } catch (Exception e) {
  16.             log.warn("close file channel " + this.fileName + " Failed. ", e);
  17.         }
  19.         return true;
  20.     } else {
  21.         log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
  22.             + " Failed. cleanupOver: " + this.cleanupOver);
  23.     }
  25.     return false;
  26. }

1> 关闭 MappedFile

第一次调用时 this.available为true，设置 available 为 false，设置第一次关闭的时间戳为当前时间戳，调用 release()释放资源，只有在引用次数小于 1 的时候才会释放资源,如果引用次数大于 0，判断当前时间与 firstShutdownTimestamp 的差值是否大于最大拒绝存活期intervalForcibly,如果大于等于最大拒绝存活期，将引用数减少 1000，直到引用数小于 0 释放资源

  1. public void shutdown(final long intervalForcibly) {
  2.     if (this.available) {
  3.         this.available = false;
  4.         this.firstShutdownTimestamp = System.currentTimeMillis();
  5.         this.release();
  6.     } else if (this.getRefCount() > 0) {
  7.         if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {
  8.             this.refCount.set(-1000 - this.getRefCount());
  9.             this.release();
  10.         }
  11.     }
  12. }

2> 判断是否清理完成

是否清理完成的标准是引用次数小于等于 0 并且清理完成标记 cleanupOver 为 true

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

3> 关闭文件通道 fileChannel

this.fileChannel.close();