RocketMQ刷盘流程

前言
这里推荐两个看源码较好用的快捷键，可以提高源码阅读效率（win10系统）：
1. ctrl + alt + h：查看方法的调用链
2. ctrl + h：查看类的继承关系和接口实现关系

简介

消息存储完成后，会被操作系统持久化到磁盘，也就是刷盘。
RocketMQ 支持两种刷盘方式，根据 DefaultMessageStore.getMessageStoreConfig().getFlushDiskType() 获取到的 FlushDiskType 来判定是同步刷盘或是异步刷盘，若 FlushDiskType 为 SYNC_FLUSH，则表示同步刷盘；若为 ASYNC_FLUSH，则为异步刷盘。 RocketMQ 默认为异步刷盘，且若想使用同步刷盘，则需调用 MessageStoreConfig.setFlushDiskType() 更改 FlushDiskType 为 SYNC_FLUSH。

RocketMQ 消息刷盘机制大致分以下两部分介绍：
1. 刷盘服务线程的类型、创建和启动
2. 消息刷盘的逻辑

刷盘服务线程的类型和创建、启动

刷盘服务线程类型

消息刷盘线程分别由三个类实现：GroupCommitService、FlushRealTimeService、CommitRealTimeService，其中 GroupCommitService 负责同步刷盘服务，FlushRealTimeService 负责异步刷盘服务，CommitRealTimeService 负责异步转存服务。这三种类都是CommitLog的内部类，且都继承自 FlushCommitLogService 类。

刷盘服务线程何时创建

我们先来看看调用链：BrokerStartup.createBrokerController() -> BrokerController.initialize() -> DefaultMessageStore.DefaultMessageStore() -> CommitLog.CommitLog()，到构造CommitLog对象的时候停止，看一下CommitLog的构造函数：

public CommitLog(final DefaultMessageStore defaultMessageStore) {this.mappedFileQueue = new MappedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(), defaultMessageStore.getAllocateMappedFileService());this.defaultMessageStore = defaultMessageStore;if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {// 同步刷盘this.flushCommitLogService = new GroupCommitService();} else {// 异步刷盘this.flushCommitLogService = new FlushRealTimeService();}// 异步转存this.commitLogService = new CommitRealTimeService();this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());batchEncoderThreadLocal = new ThreadLocal<MessageExtBatchEncoder>() {@Overrideprotected MessageExtBatchEncoder initialValue() {return new MessageExtBatchEncoder(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());}};this.putMessageLock = defaultMessageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();}

可以看出，CommitLog 对象在 Broker 启动时的 BrokerStartup.createBrokerController() 方法中经过一系列调用创建，在创建 CommitLog 对象的时候，会根据 DefaultMessageStore.getMessageStoreConfig().getFlushDiskType() 值决定创建 GroupCommitService 或 FlushRealTimeService 对象，创建的对象由 FlushCommitLogService 类型的 flushCommitLogService 引用，也就是说，RocketMQ 每次启动仅能支持一种刷盘方式：同步或异步，而不同时支持同步和异步刷盘方式。同时会创建 CommitRealTimeService 对象，该对象由 FlushCommitLogService 类型的 commitLogService 引用

刷盘服务线程何时启动

还是看调用链：BrokerStartup.main() -> start() -> BrokerController.start() -> DefaultMessageStore.start() -> CommitLog.start() ，可以知道，在创建完 BrokerController 对象后启动 BrokerController 时，会调用 CommitLog.start() 方法，我们看一下这个方法

public void start() {this.flushCommitLogService.start();if (defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {this.commitLogService.start();}
}

这里可以看出，进入该方法后，flushCommitLogService 线程会被启动。同时，根据 defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable() 是否开启内存缓存池（默认不开启）决定是否启动 commitLogService 线程

小结

1. 到这里我们知道了，刷盘线程有三种，分别提供同步刷盘、异步刷盘和异步转存服务。
2. 其中，默认使用的是异步刷盘，同时默认不使用异步转存服务（即 transientStorePoolEnable 默认为 false），也就是说，RocketMQ 默认仅创建和启动 FlushRealTimeService 线程。
3. 此外，我们还知道了，刷盘服务线程在创建BrokerController的时候创建，在启动BrokerController的时候被启动。
4. 当内存缓存池 TransientStorePool 可用时（（即 transientStorePoolEnable 为 true），消息会先提交到 TransientStorePool 中的 WriteBuffer 内部，再提交到 MappedFile 的 FileChannle 中，此时异步刷盘服务就是 CommitRealTimeService。
接下来，我们来看一下线程是如何刷盘的。

消息刷盘的逻辑

消息刷盘的方法 -- handleDiskFlush()

在 CommitLog.putMessage() 方法中，会完成消息写到内存中的任务（这个过程不展开说了，需要了解的读者可以到这篇博文来查看），在该方法的末尾，会调用 handleDiskFlush() ，部分代码如下：

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {// 消息存储到内存的过程（省略）if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {this.defaultMessageStore.unlockMappedFile(unlockMappedFile);}PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);// StatisticsstoreStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());// 刷盘，消息持久化handleDiskFlush(result, putMessageResult, msg);// 主从同步handleHA(result, putMessageResult, msg);return putMessageResult;
}

在 CommitLog.handleDiskFlush() 方法中，会根据 flushDiskType 值来决定是同步刷盘还是异步刷盘，下面分别进行介绍。

同步刷盘

同步刷盘大致可以分为两部分：
1. 构造写消息请求并唤醒刷盘线程
2. 刷盘线程执行run()代码体进行消息刷盘

1. 构造写消息请求并唤醒刷盘线程

下面来看看 CommitLog.handleDiskFlush() 方法，我们先看同步刷盘的逻辑：在该部分代码中主要做了以下四件事（结合代码）：
@1：根据 getFlushDiskType() 方法获得的 flushDiskType 变量的值判断是否使用同步刷盘，若是（即该变量值为 SYNC_FLUSH），则获取 flushCommitLogService 同步刷盘线程
@2：构造刷盘请求，该请求为一个 GroupCommitRequest 对象
@3：将刷盘请求放入线程（即 GroupCommitService 对象）的写请求队列
@4：同步等待获取结果

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {// Synchronization flushif (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {final GroupCommitService service = (GroupCommitService)  this.flushCommitLogService;      //@1 if (messageExt.isWaitStoreMsgOK()) {GroupCommitRequest request = new GroupCommitRequest  (result.getWroteOffset() + result.getWroteBytes());   //@2service.putRequest(request);   //@3CompletableFuture<PutMessageStatus> flushOkFuture = request.future();   //@4PutMessageStatus flushStatus = null;try {flushStatus = flushOkFuture.get(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),TimeUnit.MILLISECONDS);} catch (InterruptedException | ExecutionException | TimeoutException e) {//flushOK=false;}if (flushStatus != PutMessageStatus.PUT_OK) {log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags()+ " client address: " + messageExt.getBornHostString());putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);}}else {service.wakeup();}}// 异步刷盘
}

接下来进入 CommitLog.GroupCommitService.putRequest() 方法，可知该方法做了如下两件事（代码注释）：

public synchronized void putRequest(final GroupCommitRequest request) {synchronized (this.requestsWrite) {// 1. 将刷盘请求添加进写请求队列this.requestsWrite.add(request);}// 2. 唤醒刷盘线程处理请求this.wakeup();
}

再看看 ServiceThread.wakeup() 方法，查看代码可知，该方法用于唤醒刷盘线程，以实现主线程（存储消息线程）和刷盘线程间的协调

public void wakeup() {// hasNotified默认为false，因此compareAndSet方法返回true，同时会将hasNotified修改为trueif (hasNotified.compareAndSet(false, true)) {// waitPoint 是 CountDownLatch2 类型对象，count 值为1，因此 countDown 执行后会唤醒刷盘线程waitPoint.countDown(); // notify}
}

可以知道，到这里，刷盘线程被唤醒了。接下来，刷盘线程将执行相应的 run()方法，完成自己的工作

2. 刷盘线程执行 run()代码体进行消息刷盘

我们查看同步刷盘线程的 CommitLog.GroupCommitService.run() 方法，可以知道线程的执行代码：

public void run() {CommitLog.log.info(this.getServiceName() + " service started");// 刷盘线程是否停止while (!this.isStopped()) {try {this.waitForRunning(10);   // @1this.doCommit();   // @2} catch (Exception e) {CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);}}// Under normal circumstances shutdown, wait for the arrival of the// request, and then flush// 若线程已经停止，则在等待一段时间后，将剩余的刷盘请求进行刷盘try {Thread.sleep(10);} catch (InterruptedException e) {CommitLog.log.warn("GroupCommitService Exception, ", e);}synchronized (this) {this.swapRequests();}this.doCommit();CommitLog.log.info(this.getServiceName() + " service end");
}

该方法是线程的主要执行逻辑，解释如下：

@1：this.waitForRunning(10); 语句，进入该方法查看，可知该方法交换了读写队列，这个操作有以下好处：可实现读写分离，使刷盘线程在进行刷盘（读读消息队列）的时候，存储消息的线程仍然可以将写请求添加到写消息队列，避免产生锁竞争。

protected void waitForRunning(long interval) {// hasNotified在上面的wakeup()唤醒刷盘线程后已变为true，因此该if语句返回true，同时hasNotified变回falseif (hasNotified.compareAndSet(true, false)) {// 交换读写队列this.onWaitEnd();return;}//entry to waitwaitPoint.reset();try {waitPoint.await(interval, TimeUnit.MILLISECONDS);} catch (InterruptedException e) {log.error("Interrupted", e);} finally {hasNotified.set(false);this.onWaitEnd();}
}

@2：this.doCommit(); 语句，该方法是线程刷盘过程的主要实现，解释如下：
Ⅰ：对读请求队列加锁
Ⅱ：遍历读请求队列，取出读请求队列中的刷盘请求。从这里可以看出，RocketMQ 是一次处理一批刷盘请求
Ⅲ：因为消息可能分别存在了两个 mappedFile 中，因此需要至少刷盘两次。每次刷盘会更新 flushOK 变量，该变量值由刷盘后消息的偏移量和请求消息的偏移量的比较来确定，用于判断此次刷盘是否结束。若尚未结束，则进行第二次刷盘。刷盘调用的是 MappedFileQueue.flush() 方法，此时开始真正刷盘，具体细节放到了后面
Ⅳ：唤醒等待刷盘结果的线程

private void doCommit() {synchronized (this.requestsRead) {   // Ⅰif (!this.requestsRead.isEmpty()) {   // 走这里for (GroupCommitRequest req : this.requestsRead) {   // Ⅱ// There may be a message in the next file, so a maximum of// two times the flushboolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();for (int i = 0; i < 2 && !flushOK; i++) {   // Ⅲ// 调用MappedFileQueue的flush()方法刷盘CommitLog.this.mappedFileQueue.flush(0);flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();}req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);   // Ⅳ}long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();if (storeTimestamp > 0) {CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);}this.requestsRead.clear();} else {// Because of individual messages is set to not sync flush, it// will come to this processCommitLog.this.mappedFileQueue.flush(0);}}
}

异步刷盘

起点仍然是CommitLog.handleDiskFlush()方法

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {// 同步刷盘过程（省略）// Asynchronous flushelse {if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {// 若没有开启内存缓存池，则唤醒 flushCommitLogServiceflushCommitLogService.wakeup();} else {// 否则唤醒 commitLogServicecommitLogService.wakeup();}}
}

FlushRealTimeService 在启动后，会在死循环中周期性的进行刷盘操作

FlushRealTimeService.run()
该方法实现了异步刷盘的主逻辑。这里面有几个比较重要的参数，如下：
boolean flushCommitLogTimed：休眠策略，为 true 时，调用 Thread.sleep()休眠，为 false 时，调用 waitForRunning（）休眠，每次休眠时长 interval 大小，默认 false
int interval：休眠时长，也作刷盘周期，默认为 500ms
int flushPhysicQueueLeastPages：每次刷盘至少要刷多少页内容，每页大小为 4 k，默认每次要刷 4 页
int flushPhysicQueueThoroughInterval：两次刷写之间的最大时间间隔，默认 10 s

接下来结合代码解释该方法的主逻辑：
@1：若距离上次刷盘时间间隔大于 flushPhysicQueueThoroughInterval，则将 flushPhysicQueueLeastPages 设置为0，表明将所有内存缓存全部刷到文件中
@2：根据不同休眠策略，进行休眠等待，默认 flushCommitLogTimed 为 false，即默认走 @3
@3：默认走这里，调用 ServiceThread.waitForRunning()方法，休眠 interval 大小的时长
@4：与同步刷盘一样，刷盘时调用的是 MappedFileQueue.flush()方法
@5：若线程被停止了，则重试 RETRY_TIMES_OVER（默认为10）大小的次数，每次重试进行一次刷盘，直到内存中所有消息完成刷盘

从 @3 语句中我们会以为异步刷盘是每隔 500ms 刷盘一次，但结合 CommitLog.handleDiskFlush() 方法，可以知晓每异步写入一条消息，都会触发 flushCommitLogService.wakeup() 直接中断 this.waitForRunning(interval)。因此异步刷盘并非想当然的每隔500ms 刷一次盘。而是如果没有新的消息写入，会休眠 500ms，但收到了新的消息后，可以被唤醒，做到消息及时被刷盘，而不是一定要等 500 ms。

public void run() {CommitLog.log.info(this.getServiceName() + " service started");while (!this.isStopped()) {boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();int flushPhysicQueueThoroughInterval =CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();boolean printFlushProgress = false;// Print flush progresslong currentTimeMillis = System.currentTimeMillis();if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {   // @1this.lastFlushTimestamp = currentTimeMillis;flushPhysicQueueLeastPages = 0;printFlushProgress = (printTimes++ % 10) == 0;}try {if (flushCommitLogTimed) {   // @2Thread.sleep(interval);} else {this.waitForRunning(interval);   // @3}if (printFlushProgress) {this.printFlushProgress();}long begin = System.currentTimeMillis();CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);   // @4long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();if (storeTimestamp > 0) {CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);}long past = System.currentTimeMillis() - begin;if (past > 500) {log.info("Flush data to disk costs {} ms", past);}} catch (Throwable e) {CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);this.printFlushProgress();}}// Normal shutdown, to ensure that all the flush before exitboolean result = false;for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {   // @5result = CommitLog.this.mappedFileQueue.flush(0);CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));}this.printFlushProgress();CommitLog.log.info(this.getServiceName() + " service end");
}

消息最终刷盘

无论是同步刷盘还是异步刷盘，在主流程的逻辑处理完后，最终都是调用 MappedFileQueue 的 flush()方法进行消息刷盘

MappedFileQueue.flush()
该方法的主要逻辑解释如下（结合代码）：
@1：flushedWhere 记录了最后一条被刷到文件的内容的全局物理偏移量。所以此次刷盘就要根据偏移量，找到本次要刷盘的起始点位于哪个mappedFile，该 mappedFile 存储在 CopyOnWriteArrayList 类型的列表里
@2：调用 MappedFile.flush()方法刷盘
@3：更新 flushedWhere 值

public boolean flush(final int flushLeastPages) {boolean result = true;MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, this.flushedWhere == 0);   // @1if (mappedFile != null) {long tmpTimeStamp = mappedFile.getStoreTimestamp();int offset = mappedFile.flush(flushLeastPages);   // @2long where = mappedFile.getFileFromOffset() + offset;result = where == this.flushedWhere;this.flushedWhere = where;   // @3if (0 == flushLeastPages) {this.storeTimestamp = tmpTimeStamp;}}return result;
}

MappedFile.flush()
同样结合代码分析主逻辑如下：
@1：校验是否满足刷盘条件，该方法根据 flushLeastPages 的值，有两种处理逻辑
1. 若 flushLeastPages 值为0，对比 wrotePosition 和 flushedPosition 的值，若 flushedPosition > wrotePosition，则返回 true
2. 若 flushLeastPages 值大于0，则判断当前剩余未刷盘内容长度，是否超过最小刷盘长度 flushLeastPages，若超过，则返回 true，避免不必要的刷盘操作。
@2：校验 mappedFile 是否还能用
@3：因为默认不开启内存缓存池（即 transientStorePoolEnable 默认为 false），所以将使用 @4 中的 mappedByteBuffer 存储消息
@4：最终调用 MappedByteBuffer.force()方法刷盘

public int flush(final int flushLeastPages) {if (this.isAbleToFlush(flushLeastPages)) {   // @1if (this.hold()) {   // @2int value = getReadPosition();try {//We only append data to fileChannel or mappedByteBuffer, never both.if (writeBuffer != null || this.fileChannel.position() != 0) {   // @3this.fileChannel.force(false);} else {this.mappedByteBuffer.force();   // @4}} catch (Throwable e) {log.error("Error occurred when force data to disk.", e);}this.flushedPosition.set(value);this.release();} else {log.warn("in flush, hold failed, flush offset = " + this.flushedPosition.get());this.flushedPosition.set(getReadPosition());}}return this.getFlushedPosition();
}

至此消息最终刷到磁盘中，一次同步或异步刷盘结束