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

// org.apache.rocketmq.store.TransientStorePool
public class TransientStorePool {
    // 存儲池大小，默認是5
    private final int poolSize;
    // CommitLog MappedFile文件大小，默認1GB
    private final int fileSize;
    // 默認存5個ByteBuffer
    private final Deque<ByteBuffer> availableBuffers;
    // 消息存儲配置
    private final MessageStoreConfig storeConfig;
		// TransientStorePool初始化
    public void init() {
        // 默認是5
        for (int i = 0; i < poolSize; i++) {
            // 分配1GB的直接內存
            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);
        }
    }
    // 歸還緩沖
    public void returnBuffer(ByteBuffer byteBuffer) {
        // 修改position和limit，"清空"緩沖
        byteBuffer.position(0);
        byteBuffer.limit(fileSize);
      	// 緩沖入隊
        this.availableBuffers.offerFirst(byteBuffer);
    }
    // 向TransientStorePool借緩沖
    public ByteBuffer borrowBuffer() {
      	// 緩沖出隊
        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的內存映射(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/*文件長度-當前寫位置，可以寫的長度*/,
                    (MessageExtBrokerInner) messageExt, putMessageContext);
        } // ...如果是批量消息
        return result;
    }
}

private boolean mmapOperation() {
  // ...
  if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
      try {
          mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
        	// 初始化mappedFile會向TransientStorePool"借"一個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);
    // 提交復制請求
    CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
    // 合并提交刷盤請求和提交復制請求結果
    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 = 當前寫入偏移量 + 當前消息寫入大小
            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 {
            // 同步刷盤，但是不需要等待刷盤結果，那么喚醒同步刷盤線程
            service.wakeup();
            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        }
    }
    // 異步刷盤
    else {
        // 是否啟動了堆外緩存
        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            // 如果沒有啟動堆外緩存，則喚醒異步刷盤服務 flushRealTimeService
            flushCommitLogService.wakeup();
        } else  {
            // 如果啟動了堆外緩存，則喚醒異步轉存服務CommitRealTimeService
            commitLogService.wakeup();
        }
        return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
    }
}