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

RocketMQ CommitLog 详解

commitlog 目录主要存储消息，为了保证性能，顺序写入，每一条消息的长度都不相同，每条消息的前面四个字节存储该条消息的总长度，每个文件大小默认为 1G，文件的命名是以 commitLog 起始偏移量命名的，可以通过修改 broker 配置文件中 mappedFileSizeCommitLog 属性改变文件大小

1、获取最小偏移量

org.apache.rocketmq.store.CommitLog#getMinOffset

  1. public long getMinOffset() {
  2.     MappedFile mappedFile = this.mappedFileQueue.getFirstMappedFile();
  3.     if (mappedFile != null) {
  4.         if (mappedFile.isAvailable()) {
  5.             return mappedFile.getFileFromOffset();
  6.         } else {
  7.             return this.rollNextFile(mappedFile.getFileFromOffset());
  8.         }
  9.     }
  11.     return -1;
  12. }

获取目录下的第一个文件

  1. public MappedFile getFirstMappedFile() {
  2.     MappedFile mappedFileFirst = null;
  4.     if (!this.mappedFiles.isEmpty()) {
  5.         try {
  6.             mappedFileFirst = this.mappedFiles.get(0);
  7.         } catch (IndexOutOfBoundsException e) {
  8.             //ignore
  9.         } catch (Exception e) {
  10.             log.error("getFirstMappedFile has exception.", e);
  11.         }
  12.     }
  14.     return mappedFileFirst;
  15. }

如果该文件可用返回文件的起始偏移量,否则返回下一个文件的 起始偏移量

  1. public long rollNextFile(final long offset) {
  2.     int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
  3.     return offset + mappedFileSize - offset % mappedFileSize;
  4. }

2、根据偏移量和消息长度查找消息

org.apache.rocketmq.store.CommitLog#getMessage

  1. public SelectMappedBufferResult getMessage(final long offset, final int size) {
  2.     int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
  3.     MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
  4.     if (mappedFile != null) {
  5.         int pos = (int) (offset % mappedFileSize);
  6.         return mappedFile.selectMappedBuffer(pos, size);
  7.     }
  8.     return null;
  9. }

首先获取 commitLog 文件大小,默认 1G

private int mappedFileSizeCommitLog = 1024 * 1024 * 1024;

获取偏移量所在的 MappedFile

org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)

获取第一个 MappedFile 和最后一个 MappedFile，校验偏移量是否在这两个 MappedFile 之间,计算当前偏移量所在 MappedFiles 索引值为当前偏移量的索引减去第一个文件的索引值

  1. if (firstMappedFile != null && lastMappedFile != null) {
  2.     if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
  3.         LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
  4.             offset,
  5.             firstMappedFile.getFileFromOffset(),
  6.             lastMappedFile.getFileFromOffset() + this.mappedFileSize,
  7.             this.mappedFileSize,
  8.             this.mappedFiles.size());
  9.     } else {
  10.         int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
  11.         MappedFile targetFile = null;
  12.         try {
  13.             targetFile = this.mappedFiles.get(index);
  14.         } catch (Exception ignored) {
  15.         }
  17.         if (targetFile != null && offset >= targetFile.getFileFromOffset()
  18.             && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
  19.             return targetFile;
  20.         }
  22.         for (MappedFile tmpMappedFile : this.mappedFiles) {
  23.             if (offset >= tmpMappedFile.getFileFromOffset()
  24.                 && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
  25.                 return tmpMappedFile;
  26.             }
  27.         }
  28.     }
  30.     if (returnFirstOnNotFound) {
  31.         return firstMappedFile;
  32.     }
  33. }

根据在文件内的偏移量和消息长度获取消息内容

  1. public SelectMappedBufferResult selectMappedBuffer(int pos, int size) {
  2.     int readPosition = getReadPosition();
  3.     if ((pos + size) <= readPosition) {
  4.         if (this.hold()) {
  5.             ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
  6.             byteBuffer.position(pos);
  7.             ByteBuffer byteBufferNew = byteBuffer.slice();
  8.             byteBufferNew.limit(size);
  9.             return new SelectMappedBufferResult(this.fileFromOffset + pos, byteBufferNew, size, this);
  10.         } else {
  11.             log.warn("matched, but hold failed, request pos: " + pos + ", fileFromOffset: "
  12.                 + this.fileFromOffset);
  13.         }
  14.     } else {
  15.         log.warn("selectMappedBuffer request pos invalid, request pos: " + pos + ", size: " + size
  16.             + ", fileFromOffset: " + this.fileFromOffset);
  17.     }
  19.     return null;
  20. }

3、Broker 正常停止文件恢复

org.apache.rocketmq.store.CommitLog#recoverNormally

首先查询消息是否验证 CRC

boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();

从倒数第 3 个文件开始恢复，如果不足 3 个文件，则从第一个文件开始恢复

  1. int index = mappedFiles.size() - 3;
  2. if (index < 0)
  3.     index = 0;

循环遍历 CommitLog 文件，每次取出一条消息

DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);

如果查找结果为 true 并且消息的长度大于 0，表示消息正确，mappedFileOffset 指针向前移动本条消息的长度；

  1. if (dispatchRequest.isSuccess() && size > 0) {
  2.     mappedFileOffset += size;
  3. }

如果查找结果为 true 并且结果等于 0，表示已到文件 的末尾，如果还有下一个文件，则重置 processOffset、mappedOffset 并重复上述步骤，否则跳出循环；

  1. else if (dispatchRequest.isSuccess() && size == 0) {
  2.     index++;
  3.     if (index >= mappedFiles.size()) {
  4.         // Current branch can not happen
  5.         log.info("recover last 3 physics file over, last mapped file " + mappedFile.getFileName());
  6.         break;
  7.     } else {
  8.         mappedFile = mappedFiles.get(index);
  9.         byteBuffer = mappedFile.sliceByteBuffer();
  10.         processOffset = mappedFile.getFileFromOffset();
  11.         mappedFileOffset = 0;
  12.         log.info("recover next physics file, " + mappedFile.getFileName());
  13.     }
  14. }

如果查找结果为 false，则表示消息没有填满该文件，跳出循环，结束遍历

  1. else if (!dispatchRequest.isSuccess()) {
  2.     log.info("recover physics file end, " + mappedFile.getFileName());
  3.     break;
  4. }

更新 committedPosition 和 flushedWhere 指针

  1. this.mappedFileQueue.setFlushedWhere(processOffset);
  2. this.mappedFileQueue.setCommittedWhere(processOffset);

删除 offset 之后的所有文件。遍历目录下面的所有文件，如果文件尾部偏移量小于 offset 则跳过该文件，如果尾部的偏移量大于 offset，则进一步比较 offset 与文件的开始偏移量，如果 offset 大于文件的开始偏移量，说明当前文件包含了有效偏移量，设置 MappedFile 的 flushPosition 和 commitedPosition。

如果 offset 小于文件的开始偏移量，说明该文件是有效文件后面创建的，调用 MappedFile#destroy()方法释放资源

  1. if (fileTailOffset > offset) {
  2.     if (offset >= file.getFileFromOffset()) {
  3.         file.setWrotePosition((int) (offset % this.mappedFileSize));
  4.         file.setCommittedPosition((int) (offset % this.mappedFileSize));
  5.         file.setFlushedPosition((int) (offset % this.mappedFileSize));
  6.     } else {
  7.         file.destroy(1000);
  8.         willRemoveFiles.add(file);
  9.     }
  10. }

释放资源需要关闭 MappedFile 和文件通道 fileChannel

  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. }

判断maxPhyOffsetOfConsumeQueue是否大于 processOffset，如果大于，需要删除 ConsumeQueue 中 processOffset 之后的数据

  1. if (maxPhyOffsetOfConsumeQueue >= processOffset) {
  2.     log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
  3.     this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
  4. }
  1. public void truncateDirtyLogicFiles(long phyOffset) {
  2.     ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> tables = DefaultMessageStore.this.consumeQueueTable;
  4.     for (ConcurrentMap<Integer, ConsumeQueue> maps : tables.values()) {
  5.         for (ConsumeQueue logic : maps.values()) {
  6.             logic.truncateDirtyLogicFiles(phyOffset);
  7.         }
  8.     }
  9. }