public class CommitLog {
  // 如果是同步刷盤，則是GroupCommitService。如果是異步刷盤則是FlushRealTimeService
  // 默認(rèn)是異步刷盤，因此是CommitLog$FlushRealTimeService
  private final FlushCommitLogService flushCommitLogService;
  // 開啟TransientStorePoolEnable時(shí)使用CommitRealTimeService
  private final FlushCommitLogService commitLogService;
	// 構(gòu)造函數(shù)
  public CommitLog(final DefaultMessageStore defaultMessageStore) {
      // 默認(rèn)是異步刷盤，因此這里是false
      if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
          this.flushCommitLogService = new GroupCommitService();
      } else {
          this.flushCommitLogService = new FlushRealTimeService();
      }
      this.commitLogService = new CommitRealTimeService();
      // 消息回調(diào)
      this.appendMessageCallback = new DefaultAppendMessageCallback();
      flushDiskWatcher = new FlushDiskWatcher();
  }
}

// org.apache.rocketmq.store.TransientStorePool
public class TransientStorePool {
    // 存儲池大小，默認(rèn)是5
    private final int poolSize;
    // CommitLog MappedFile文件大小，默認(rèn)1GB
    private final int fileSize;
    // 默認(rèn)存5個(gè)ByteBuffer
    private final Deque<ByteBuffer> availableBuffers;
    // 消息存儲配置
    private final MessageStoreConfig storeConfig;
		// TransientStorePool初始化
    public void init() {
        // 默認(rèn)是5
        for (int i = 0; i < poolSize; i++) {
            // 分配1GB的直接內(nèi)存
            ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);
            final long address = ((DirectBuffer) byteBuffer).address();
            Pointer pointer = new Pointer(address);
            LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));
            // 生成的緩存保存到隊(duì)列中
            availableBuffers.offer(byteBuffer);
        }
    }
    // 歸還緩沖
    public void returnBuffer(ByteBuffer byteBuffer) {
        // 修改position和limit，"清空"緩沖
        byteBuffer.position(0);
        byteBuffer.limit(fileSize);
      	// 緩沖入隊(duì)
        this.availableBuffers.offerFirst(byteBuffer);
    }
    // 向TransientStorePool借緩沖
    public ByteBuffer borrowBuffer() {
      	// 緩沖出隊(duì)
        ByteBuffer buffer = availableBuffers.pollFirst();
        return buffer;
    }
}

// org.apache.rocketmq.store.MappedFile#appendMessagesInner
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb,
        PutMessageContext putMessageContext) {
    // 如果寫文件位置小于文件size
    if (currentPos < this.fileSize) {
        // 如果writeBuffer不空，則獲取writeBuffer的淺拷貝，否則獲取MappedFile的內(nèi)存映射(MappedByteBuffer)的淺拷貝
        ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
        byteBuffer.position(currentPos);
        AppendMessageResult result;
        // 如果是單條消息
        if (messageExt instanceof MessageExtBrokerInner) {
            result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos/*文件長度-當(dāng)前寫位置，可以寫的長度*/,
                    (MessageExtBrokerInner) messageExt, putMessageContext);
        } // ...如果是批量消息
        return result;
    }
}

private boolean mmapOperation() {
  // ...
  if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
      try {
          mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
        	// 初始化mappedFile會向TransientStorePool"借"一個(gè)writeBuffer
          mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
      } catch (RuntimeException e) {
          mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
      }
  } else {
    	// 創(chuàng)建MappedFile，沒有writeBuffer
      mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
  }
  // ...
}

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
    //... 保存消息
    result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
    // ...
    // 提交刷盤請求
    CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
    // 提交復(fù)制請求
    CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
    // 合并提交刷盤請求和提交復(fù)制請求結(jié)果
    return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
        if (flushStatus != PutMessageStatus.PUT_OK) {
            putMessageResult.setPutMessageStatus(flushStatus);
        }
        if (replicaStatus != PutMessageStatus.PUT_OK) {
            putMessageResult.setPutMessageStatus(replicaStatus);
        }
        return putMessageResult;
    });
}

public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {
    // 同步刷盤
    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
        // 獲取同步刷盤Service
        final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
        if (messageExt.isWaitStoreMsgOK()) {
            // 創(chuàng)建GroupCommitRequest 刷盤偏移量nextOffset = 當(dāng)前寫入偏移量 + 當(dāng)前消息寫入大小
            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
                    this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
            // 向刷盤監(jiān)視器(flushDistWatch)提交刷盤請求
            flushDiskWatcher.add(request);
            // 提交刷盤請求，并且喚醒同步刷盤線程
            service.putRequest(request);
            return request.future();
        } else {
            // 同步刷盤，但是不需要等待刷盤結(jié)果，那么喚醒同步刷盤線程
            service.wakeup();
            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        }
    }
    // 異步刷盤
    else {
        // 是否啟動(dòng)了堆外緩存
        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            // 如果沒有啟動(dòng)堆外緩存，則喚醒異步刷盤服務(wù) flushRealTimeService
            flushCommitLogService.wakeup();
        } else  {
            // 如果啟動(dòng)了堆外緩存，則喚醒異步轉(zhuǎn)存服務(wù)CommitRealTimeService
            commitLogService.wakeup();
        }
        return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
    }
}