RocketMQ Broker消息如何刷盘源码解析

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 {
    	// 创建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()) {
            // 创建GroupCommitRequest 刷盘偏移量nextOffset = 当前写入偏移量 + 当前消息写入大小
            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
                    this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
            // 向刷盘监视器(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);
    }
}