RocketMQ源码分析之消息刷盘

中华儿女多奇志，不爱无码爱代码

一：前言

在上篇文章《RocketMQ源码分析之消息存储》中，我们分析了消息是如何从 Broker 最终存储到MappedFile 内存缓冲区中的，但是此时消息存储的任务并没有完成，因为消息还没有刷盘，即存储到文件中，本篇我们就来看看RocketMQ是如何进行消息刷盘的。

二：刷盘策略

CommitLog在初始化的时候，会根据配置，启动两种不同的刷盘服务。

if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
  this.flushCommitLogService = new GroupCommitService();
} else {
  this.flushCommitLogService = new FlushRealTimeService();
}

1：同步刷盘

同步的意思就是说当消息追加到内存后，就立即刷到文件中存储。

2：异步刷盘

当消息追加到内存中，并不是理解刷到文件中，而是在后台任务中进行异步操作。

RocketMQ默认采用异步刷盘策略。

当CommitLog在putMessage（）中收到MappedFile成功追加消息到内存的结果后，便会调用handleDiskFlush（）方法进行刷盘，将消息存储到文件中。handleDiskFlush（）便会根据两种刷盘策略，调用不同的刷盘服务。

三：同步刷盘

同步刷盘的服务为GroupCommitService，主要逻辑如下：

（1）：handleDiskFlush（）中提交刷盘请求

final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;

GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);

（2）：同步等待刷盘结果，刷盘失败也会标志消息存储失败，返回 FLUSH_DISK_TIMEOUT

boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
if (!flushOK) {
  log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags()
      + " client address: " + messageExt.getBornHostString());
  putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
}

进行同步刷盘的服务为 GroupCommitService，当请求被提交给GroupCommitService后，GroupCommitService并不是立即处理，而是先放到内部的一个请求队列中，并利用waitPoint通知新请求到来。

public synchronized void putRequest(final GroupCommitRequest request) {
      synchronized (this.requestsWrite) {
          this.requestsWrite.add(request);
      }
      if (hasNotified.compareAndSet(false, true)) {
          waitPoint.countDown(); // notify
      }
}

当 GroupCommitService 被唤醒后，便会将 requestsWrite 中的请求交换到 requestsRead中，避免产生锁竞争。

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

GroupCommitService 在启动后会在死循环中调用doCommit（）方法，而doCommit（）则不断遍历requestsRead中的请求，进行处理：

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 flush
                boolean flushOK = false;
                for (int i = 0; i < 2 && !flushOK; i++) {
                    flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();

                    if (!flushOK) {
                        CommitLog.this.mappedFileQueue.flush(0);
                    }
                }

                req.wakeupCustomer(flushOK);
            }

            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 process
            CommitLog.this.mappedFileQueue.flush(0);
        }
    }
}

可见这里最终调用了CommitLog.this.mappedFileQueue.flush(0) 来进行刷盘。

同步刷盘的任务虽然也是在异步线程中执行，但是消息存储的主流程中会同步等待刷盘结果，所以本质上还是同步操作。

四：异步刷盘

同步刷盘的服务为FlushRealTimeService，不过当内存缓存池TransientStorePool 可用时，消息会先提交到TransientStorePool 中的WriteBuffer内部，再提交到MappedFile的FileChannle中，此时异步刷盘服务就是 CommitRealTimeService，它继承自 FlushRealTimeService。

我们别管那么多，先看看FlushRealTimeService中的主要逻辑吧：

（1）：handleDiskFlush（）中直接唤醒异步刷盘服务

 flushCommitLogService.wakeup();

（2）：FlushRealTimeService 在启动后，会在死循环中周期性的进行刷盘操作，主要逻辑如下。

while (!this.isStopped()) {
    // 休眠策略，为 true 时，调用 Thread.sleep()休眠，为false时，调用wait（）休眠，默认 false
    boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();

    // 获取刷盘周期，默认为 500 ms
    int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
    // 每次刷盘至少要刷多少页内容，每页大小为 4 k，默认每次要刷 4 页
    int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();
    // 两次刷写之间的最大时间间隔，默认 10 s
    int flushPhysicQueueThoroughInterval =
        CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();

    boolean printFlushProgress = false;

    // Print flush progress
    long currentTimeMillis = System.currentTimeMillis();
    // 判断当前时间距离上次刷盘时间是否已经超出设置的两次刷盘最大间隔
    if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {
        this.lastFlushTimestamp = currentTimeMillis;
        // 如果已经超时，则将flushPhysicQueueLeastPages设置为0，表明将所有内存缓存全部刷到文件中
        flushPhysicQueueLeastPages = 0;
        printFlushProgress = (printTimes++ % 10) == 0;
    }

    try {
        // 根据不同休眠策略，进行休眠等待
        if (flushCommitLogTimed) {
            Thread.sleep(interval);
        } else {
            this.waitForRunning(interval);
        }

        if (printFlushProgress) {
            this.printFlushProgress();
        }

        long begin = System.currentTimeMillis();

        // 休眠结束，开始执行刷盘操作
        CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);
        long 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();
    }
}

通过上面这段逻辑可知，异步刷盘就在异步线程中，周期性的将内存缓冲区的内容刷到文件中，在消息主流程中，只会唤醒异步刷盘线程，而不会同步等待刷盘结果，所以称为异步刷盘。

五：MappedFile的刷盘

无论是上面哪种刷盘策略，最终都调用了下面这个方法进行刷盘：

CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);

是时候看看mappedFileQueue.flush（）中做了什么了。

1：从mappedFileQueue保存的所有MappedFile中，找出所要刷盘的MappedFile

MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, this.flushedWhere == 0);

flushedWhere 记录了最后一条被刷到文件的内容的全局物理偏移量。所以此次刷盘就要根据偏移量，找到本次要刷盘的起始点位于哪个MappedFile。

2：如果找到了对应的MappedFile，则对该MappedFile中的内容执行刷盘操作，并更新flushedWhere。

if (mappedFile != null) {
    long tmpTimeStamp = mappedFile.getStoreTimestamp();
    int offset = mappedFile.flush(flushLeastPages);
    long where = mappedFile.getFileFromOffset() + offset;
    result = where == this.flushedWhere;
    this.flushedWhere = where;
    if (0 == flushLeastPages) {
        this.storeTimestamp = tmpTimeStamp;
    }
}

刷盘的终极目的地就在MappedFile的flush（）方法中，具体也分为下面几步：

1：判断是否满足刷盘条件

if (this.isAbleToFlush(flushLeastPages))

isAbleToFlush（）其实就是判断当前剩余未刷盘内容长度，是否超过最小刷盘长度：flushLeastPages，避免不必要的刷盘操作。

private boolean isAbleToFlush(final int flushLeastPages) {
    int flush = this.flushedPosition.get();
    int write = getReadPosition();

    if (this.isFull()) {
        return true;
    }

    if (flushLeastPages > 0) {
        return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
    }

    return write > flush;
}

2：如果满足刷盘条件，则将内存中的内容刷到文件中。

// 如果writeBuffer不为空，则表明消息是先提交到writeBuffer中，已经从writeBuffer提交到fileChannel，直接调用fileChannel.force（）
if (writeBuffer != null || this.fileChannel.position() != 0) {
    this.fileChannel.force(false);
} else {  
    // 反之，消息是直接存储在文件内存映射缓冲区mappedByteBuffer中，直接调用它的force（）即可
    this.mappedByteBuffer.force();
}

到这儿，消息就成功的从内存中存储到文件内部了。

六：总结

通过上面的分析，我们了解了RocketMQ的两种刷盘策略：

一种是类似强一致的，保证消息存储到文件中的同步策略。

一种是提交到内存中就算存储成功，在后台异步进行刷盘的异步策略。

无论是哪种策略，肯定都有自己的优点和缺点，大家可以根据自己生成环境，选择合适的刷盘策略。

关于消息刷盘就分析到这里了，后面文章我还会继续剖析 RocketMQ。

欢迎大家点个赞，关注下！

文章链接：

汪先生：RocketMQ源码分析之服务发现

汪先生：RocketMQ源码分析之消息发送

汪先生：RocketMQ源码分析之消息存储

汪先生：RocketMQ源码分析之消息刷盘

汪先生：RocketMQ源码分析之ConsumeQueue

听说喜欢点关注的同学都长得帅

RocketMQ源码分析之消息刷盘

RocketMQ源码分析之消息刷盘

一：前言

二：刷盘策略

三：同步刷盘

四：异步刷盘

五：MappedFile的刷盘

六：总结

Recommend

RocketMQ源码分析之消息存储

Jvm之用java解析class文件

JVM之创建对象源码分析

JVM同步方法之偏向锁

Tournament Seeding Placement

Colemak Switchover

Config management in rust

Three Way Charms

Impersonating kube service accounts

Kubernetes operators in rust

About Joyk