RocketMQ源码：broker 文件刷盘机制

妖狐艹你老母 2024-03-24 12:12 94阅读 0赞

`RocketMQ`的存储与读写是基于`JDK NIO`的内存映射机制（`MappedByteBuffer`）的，消息存储时首先将消息追加到内存中，再根据配置的刷盘策略在不同时间刷盘。

*  如果是`同步刷盘`，消息追加到内存后，将同步调用`MappedByteBuffer`的`force()`方法；
 *  如果是`异步刷盘`，在消息追加到内存后会立刻返回给消息发送端。

RocketMQ使用一个`单独的线程按照某一个设定的频率执行刷盘操作`。通过在broker配置文件中配置`flushDiskType`来设定刷盘方式，可选值为`ASYNC_FLUSH`（异步刷盘）、`SYNC_FLUSH`（同步刷盘），`默认为异步刷盘`。

本节以`CommitLog`文件刷盘机制为例来剖析RocketMQ的刷盘机制，`ConsumeQueue`文件、`Index`文件刷盘的实现原理与`CommitLog`刷盘机制类似。

RocketMQ处理刷盘的实现方法为`Commitlog#handleDiskFlush()`，刷盘流程作为消息发送、消息存储的 子流程。值得注意的是，`Index文件的刷盘并不是采取定时刷盘机制`，而是`每更新一次Index文 件就会将上一次的改动写入磁盘`。

## 1. 刷盘策略 ##

![format_png][]

在理解`RocketMQ`刷盘实现之前，先理解一下上图展示的刷盘的2种实现的：

1.  直接通过`内存映射文件`，通过`flush刷新到磁盘`
2.  当异步刷盘且启用了对外内存池的时候，先`write`到`writeBuffer`，然后`commit`到`Filechannel`，最后`flush`到磁盘

`CommitLog`的`asyncPutMessage`方法中可以看到在写入消息之后，调用了`submitFlushRequest`方法执行刷盘策略：

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
        ...
        // 获取最后一个 MappedFile
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
        ...
        try {
            ...
            // todo 往mappedFile追加消息
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            ...
        } finally {
            putMessageLock.unlock();
        }
    
        ...
        // todo 消息首先进入pagecache，然后执行刷盘操作，
        CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
        ...
    }
    复制代码

刷盘有两种策略：

*  `同步刷盘`，表示消息写入到内存之后需要立刻刷到磁盘文件中。
    
    > 同步刷盘会构建`GroupCommitRequest`组提交请求并设置本次刷盘后的位置偏移量的值（写入位置偏移量+写入数据字节数），然后将请求添加到`GroupCommitService`中进行刷盘。
 *  `异步刷盘`，表示消息写入内存成功之后就返回，由MQ定时将数据刷入到磁盘中，`会有一定的数据丢失风险`。

`CommitLog#submitFlushRequest`如下：

public CompletableFuture<PutMessageStatus> submitFlushRequest(AppendMessageResult result, MessageExt messageExt) {
        // Synchronization flush 同步刷盘
        if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
            // 获取GroupCommitService
            final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
            // 是否等待
            if (messageExt.isWaitStoreMsgOK()) {
                // 构建组提交请求，传入本次刷盘后位置的偏移量：写入位置偏移量+写入数据字节数
                GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),
                        this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
                // 刷盘请求
                service.putRequest(request);
                return request.future();
            } else {
                service.wakeup();
                return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
            }
        }
        // Asynchronous flush  异步刷盘 这个就是靠os
        else {
            // 如果未使用暂存池
            if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                // 唤醒刷盘线程进行刷盘
                flushCommitLogService.wakeup();
            } else  {
                // 如果使用暂存池，使用commitLogService，先将数据写入到FILECHANNEL，然后统一进行刷盘
                commitLogService.wakeup();
            }
            // 返回结果
            return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        }
    }
    复制代码

## 2. 同步刷盘 ##

如果使用的是同步刷盘，首先获取了`GroupCommitService`，然后构建`GroupCommitRequest`组提交请求，将请求添加到`GroupCommitService`中，`GroupCommitService`用于提交刷盘数据。

### 2.1 GroupCommitRequest提交请求 ###

`GroupCommitRequest`是`CommitLog`的内部类：

*  **nextOffset**：写入`位置偏移量+写入数据字节数`，也就是本次刷盘成功后应该对应的`flush偏移量`
 *  **flushOKFuture**：刷盘结果
 *  **timeoutMillis**：刷盘的超时时间，超过超时时间还未刷盘完毕会被认为超时

public static class GroupCommitRequest {
        // 刷盘点偏移量
        private final long nextOffset;
        // 刷盘状态
        private CompletableFuture<PutMessageStatus> flushOKFuture = new CompletableFuture<>();
        private final long startTimestamp = System.currentTimeMillis();
        // 超时时间
        private long timeoutMillis = Long.MAX_VALUE;
    
        public GroupCommitRequest(long nextOffset, long timeoutMillis) {
            this.nextOffset = nextOffset;
            this.timeoutMillis = timeoutMillis;
        }
        public void wakeupCustomer(final PutMessageStatus putMessageStatus) {
            // todo 在这里调用 结束刷盘，设置刷盘状态
            this.flushOKFuture.complete(putMessageStatus);
        }
    复制代码

### 2.2 GroupCommitService处理刷盘 ###

`GroupCommitService`是`CommitLog`的内部类，从继承关系中可知它实现了Runnable接口，在run方法调用`waitForRunning`等待刷盘请求的提交，然后处理刷盘，不过这个线程是在什么时候启动的呢？

public class CommitLog {
        /**
         * GroupCommit Service
         */
        class GroupCommitService extends FlushCommitLogService {
            // ...
            // run方法
            public void run() {
                CommitLog.log.info(this.getServiceName() + " service started");
                while (!this.isStopped()) {
                    try {
                        // 等待刷盘请求的到来
                        this.waitForRunning(10);
                        // 处理刷盘
                        this.doCommit();
                    } catch (Exception e) {
                        CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
                    }
                }
                // ...
            }
        }
    }
    复制代码

#### 2.2.1 刷盘线程的启动 ####

在`BrokerController`的启动方法中，可以看到调用了`messageStore`的`start方法`，前面可知使用的是`DefaultMessageStore`，进入到`DefaultMessageStore`的start方法，它又调用了`commitLog`的start方法，在`CommitLog`的`start`方法中，启动了刷盘的线程和监控刷盘的线程：

public class BrokerController {
        public void start() throws Exception {
            if (this.messageStore != null) {
                // 启动
                this.messageStore.start();
            }
            // ...
        }
    }
    
    public class DefaultMessageStore implements MessageStore {
       /**
         * @throws Exception
         */
        public void start() throws Exception {
            // ...
            this.flushConsumeQueueService.start();
            // 调用CommitLog的启动方法
            this.commitLog.start();
            this.storeStatsService.start();
            // ...
        }
    }
    
    public class CommitLog {
        private final FlushCommitLogService flushCommitLogService; // 刷盘
        private final FlushCommitLogService commitLogService; // commitLogService
        public void start() {
            // 启动刷盘的线程
            this.flushCommitLogService.start();
            flushDiskWatcher.setDaemon(true);
            
            if (defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                this.commitLogService.start();
            }
        }
    }
    复制代码

#### 2.2.2 刷盘请求的处理 ####

既然知道了线程在何时启动的

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/6b7912520f9648d58389317ee5a0a25f.png