RocketMQ存储--同步刷盘和异步刷盘【源码笔记】

一、问题思考
二、Broker启动刷盘有关调用链
1.调用链
2.线程类关系图
三、线程类工作流程
1.堆外内存线程类CommitRealTimeService工作流程
2.同步刷盘线程类GroupCommitService工作流程
3.异步刷盘线程类FlushRealTimeService工作流程
四、消息追加与线程类的交互
1.调用链
2.同步刷盘主要代码
3.异步刷盘主要代码
五、刷盘方式示意图
1.同步刷盘示意图
2.异步刷盘未开启堆外缓存示意图
3.异步刷盘开启堆外缓存示意图
六、文章总结
七、主要源码类清单

//初始化链条
@1 BrokerStartup#main
start(createBrokerController(args));
@2 BrokerStartup#createBrokerController
final BrokerController controller = new BrokerController(...)
boolean initResult = controller.initialize();
@3 BrokerController#initialize
this.messageStore = new DefaultMessageStore(...);
@4 DefaultMessageStore#DefaultMessageStore()
this.commitLog = new CommitLog(this);
@5 CommitLog#CommitLog()
if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig()
.getFlushDiskType()) {
this.flushCommitLogService = new GroupCommitService();
} else {
this.flushCommitLogService = new FlushRealTimeService();
}
this.commitLogService = new CommitRealTimeService();
//启动链条
@6 BrokerStartup#start
controller.start();
@7 BrokerController#start()
this.messageStore.start();
@8 DefaultMessageStore#start()
this.commitLog.start();
@9 CommitLog#start()
this.flushCommitLogService.start();
if (defaultMessageStore.getMessageStoreConfig()
.isTransientStorePoolEnable()) {
this.commitLogService.start();
}

private void swapRequests() {
List<GroupCommitRequest> tmp = this.requestsWrite;
this.requestsWrite = this.requestsRead;
this.requestsRead = tmp;
}

@1 CommitLog#putMessage
//同步刷盘或者异步刷盘
handleDiskFlush(result, putMessageResult, msg);
@2 CommitLog#handleDiskFlush

if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
//等待MappedFile刷盘成功状态通过countDownLatch来控制
boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
}
}

if (!this.defaultMessageStore.getMessageStoreConfig()
.isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
commitLogService.wakeup();
}