• -------------------------------------------------------------
  • ====================================
> 技能 > 架构组件 > kafka > 源码分析RocketMQ刷盘机制

源码分析RocketMQ刷盘机制

kafka dewbay 6年前 (2019-04-12) 2247次浏览 已收录 0个评论 扫描二维码

刷盘机制支持同步刷盘和异步刷盘。为了了解其具体事项，我们以 Commitlog 的存储为例来说明RocketMQ是如何进行磁盘读写。
Comitlog#putMessage 首先，主要是将消息写入到 MappedFile,内存映射文件。然后根据刷盘策略刷写到磁盘，入口：
CommitLog#putMessage handleDiskFlush
CommitLog#handleDiskFlush

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) { // @1
// Synchronization flush
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) { // @2
final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
if (messageExt.isWaitStoreMsgOK()) {
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
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);
}
} else {
service.wakeup();
}
}
// Asynchronous flush
else { // @3
if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
commitLogService.wakeup();
}
}
}
代码@1 参数详解
AppendMessageResult result 写入到 MappedFile（内存映射文件中，bytebuffer）中的结果，具体属性包含：

wroteOffset : 下一个写入的偏移量
wroteBytes : 写入字节总长度
msgId : 消息 id
storeTimestamp : 消息存储时间，也就是写入到 MappedFile 中的时间
logicOffset : 逻辑的 consumeque 偏移量
pagecacheRT ： 写入到 MappedByteBuffer(将消息内容写入到内存映射文件中的时长)
代码@2 同步刷盘
代码@3 异步刷盘
1、同步刷盘线程
同步刷盘机制，核心实现类 CommitLog#GroupCommitService

同步刷盘核心类，竟然是一个线程，出乎我的意料。
1.1 核心属性
private volatile List requestsWrite = new ArrayList();
private volatile List requestsRead = new ArrayList();
requestsWrite : 写队列，主要用于向该线程添加刷盘任务
requestsRead ：读队列，主要用于执行特定的刷盘任务，这是是 GroupCommitService 设计的一个亮点，把读写分离，每处理完 requestsRead 中的任务，就交换这两个队列。
1.2 对外方法 putRequest，添加刷盘任务
public synchronized void putRequest(final GroupCommitRequest request) {
synchronized (this.requestsWrite) {
this.requestsWrite.add(request);
}
if (hasNotified.compareAndSet(false, true)) {
waitPoint.countDown(); // notify
}
}
该方法是很简单，就是将 GroupCommitRequest 刷盘任务放入到 requestWrite 中，就返回了，但是这个类是处理同步刷盘的，那调用方什么时候才能知道该刷盘任务已经执行了呢？
不然能说是同步刷盘呢？这又是这个类另外一个设计亮点。为了解开这个疑点，首先看一下调用方法：
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
原来奥秘在这里，放入后，request.waitForFlush,类似于 Future 模式，在这方法里进行阻塞等待。
this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS)，默认同步刷盘超时时间为 5s,那就不需要怀疑了，刷盘后，肯定会调用 countDownLatch.countDown()
GroupCommitRequest 具体类的工作机制就不细说了，其刷盘将调用的方法为：CommitLog.this.mappedFileQueue.flush(0);
在进入具体刷盘逻辑之前，我们再看下异步刷盘线程的实现。
2、异步刷盘线程
if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
commitLogService.wakeup();
}
public boolean isTransientStorePoolEnable() {
return transientStorePoolEnable && FlushDiskType.ASYNC_FLUSH == getFlushDiskType()
&& BrokerRole.SLAVE != getBrokerRole();
}
什么是 transientStorePoolEnable ，这个只能从 FlushRealTimeService 与 CommitRealTimeService 区别
2.1 FlushRealTimeService 实现机制
class FlushRealTimeService extends FlushCommitLogService {
private long lastFlushTimestamp = 0;
private long printTimes = 0;

    public void run() {
        CommitLog.log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();    // @1

            int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();    // @2
            int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();  // @3

            int flushPhysicQueueThoroughInterval =
                CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();   // @4

            boolean printFlushProgress = false;

            // Print flush progress
            long currentTimeMillis = System.currentTimeMillis();
            if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {
                this.lastFlushTimestamp = currentTimeMillis;
                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();
            }
        }

        // Normal shutdown, to ensure that all the flush before exit
        boolean result = false;
        for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
            result = 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");
    }

代码@1：flushCommitLogTimed 这个主要是等待方法，如果为 true,则使用 Thread.sleep,如果是 false 使用 waitForRunning
代码@2：interval :获取刷盘的间隔时间
代码@3：flushPhysicQueueLeastPages 每次刷盘最少需要刷新的页，（如果少于，是不是可以不放弃本次刷盘操作
代码@4：flushPhysicQueueThoroughInterval 如果上次刷新的时间+该值 小于当前时间，则改变 flushPhysicQueueLeastPages =0，并每 10 次输出异常刷新进度。
代码@5：CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages); 调用刷盘操作
代码@6：设置检测点的 StoreCheckpoint 的 physicMsgTimestamp（commitlog 文件的检测点，也就是记录最新刷盘的时间戳）
暂时不深入，在本节之后详细分析刷盘机制。
2.2 CommitRealTimeService
public void run() {
CommitLog.log.info(this.getServiceName() + ” service started”);
while (!this.isStopped()) {
int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog(); // @1

            int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages(); // @2

            int commitDataThoroughInterval =
                CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();   // @3

            long begin = System.currentTimeMillis();
            if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
                this.lastCommitTimestamp = begin;
                commitDataLeastPages = 0;
            }

            try {
                boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
                long end = System.currentTimeMillis();
                if (!result) {
                    this.lastCommitTimestamp = end; // result = false means some data committed.
                    //now wake up flush thread.
                    flushCommitLogService.wakeup();
                }

                if (end - begin > 500) {
                    log.info("Commit data to file costs {} ms", end - begin);
                }
                this.waitForRunning(interval);
            } catch (Throwable e) {
                CommitLog.log.error(this.getServiceName() + " service has exception. ", e);
            }
        }

        boolean result = false;
        for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
            result = CommitLog.this.mappedFileQueue.commit(0);
            CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
        }
        CommitLog.log.info(this.getServiceName() + " service end");
    }
}

代码@1：interval CommitRealTimeService 执行间隔
代码@2：commitDataLeastPages ：每次 commit 最少的页数
代码@3：上上面的对应，，CommitRealTimeService 与 FlushRealTimeService 不同之处，是调用的方法不一样，
FlushRealTimeService 调用 mappedFileQueue.flush，而 CommitRealTimeService 调用 commit 方法。
行文至此，我们只是了解异步刷盘，同步刷盘去线程的实现方式，接下来，是时候进入到刷盘具体逻辑，也就是 Commitlog mappedFileQueue
3、刷盘机制实现
具体实现类：MappedFileQueue
3.1 核心属性与构造方法
private static final int DELETE_FILES_BATCH_MAX = 10;

private final String storePath;

private final int mappedFileSize;

private final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();

private final AllocateMappedFileService allocateMappedFileService;

private long flushedWhere = 0;
private long committedWhere = 0;

private volatile long storeTimestamp = 0;

public MappedFileQueue(final String storePath, int mappedFileSize,
    AllocateMappedFileService allocateMappedFileService) {
    this.storePath = storePath;
    this.mappedFileSize = mappedFileSize;
    this.allocateMappedFileService = allocateMappedFileService;
}

MappedFileQueue 就是 MappedFile 的队列，也就是 MappedFile 的容器。
storePath：文件存储路径
mappedFileSize：单个 MappedFile 文件长度
mappedFiles ：mappedFile 集合
allocateMappedFileService: 创建 MappedFileService
flushedWhere :刷盘位置
committedWhere :commit 位置

我们先看一下 MappedFileQueue 在什么时候创建：
我们以 commitlog 为例：
this.mappedFileQueue = new MappedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),
defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog(), defaultMessageStore.getAllocateMappedFileService());
其中 allocateMappedFileService 为 AllocateMappedFileService
MappedFileQueue 就是 MappedFile 的队列，也就是 MappedFile 的容器。
storePath：文件存储路径
mappedFileSize：单个 MappedFile 文件长度
mappedFiles ：mappedFile 集合
allocateMappedFileService: 创建 MappedFileService
flushedWhere :刷盘位置
committedWhere :commit 位置
3.2 核心方法：load
public boolean load() {
File dir = new File(this.storePath);
File[] files = dir.listFiles();
if (files != null) {
// ascending order
Arrays.sort(files);
for (File file : files) {

            if (file.length() != this.mappedFileSize) {
                log.warn(file + "\t" + file.length()
                    + " length not matched message store config value, ignore it");
                return true;
            }

            try {
                MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);

                mappedFile.setWrotePosition(this.mappedFileSize);
                mappedFile.setFlushedPosition(this.mappedFileSize);
                mappedFile.setCommittedPosition(this.mappedFileSize);
                this.mappedFiles.add(mappedFile);
                log.info("load " + file.getPath() + " OK");
            } catch (IOException e) {
                log.error("load file " + file + " error", e);
                return false;
            }
        }
    }

    return true;
}

// 该方法主要是按顺序，创建 MappedFile，此时这个时候关注一下，初始化时 wrotePosition,flushedPosition,committedPosition 全设置为最大值，这要怎么玩呢？是否还记得启动时需要恢复 commitlog,consume,index 文件，（recover）方法，在删除无效文件时，会重置上述值

接下来，我们先梳理一下目前刷盘出现的关键属性，然后进入到刷盘机制的世界中来：
1、MappedFileQueue 与 MappedFile 的关系
可以这样认为，MappedFile 代表一个个物理文件，而 MappedFileQueue 代表由一个个 MappedFile 组成的一个连续逻辑的大文件。
并且每一个 MappedFile 的命名已该文件在整个文件序列中的偏移量来表示。
2、MappedFileQueue
flushedWhere: 整个刷新的偏移量，针对该 MappedFileQueue
committedWhere:当前提交的偏移量，针对该 MappedFileQueue commit 与 flush 的区别？
3、MappedFile
wrotePosition :当前待写入位置
committedPosition 提交位置
flushedPosition 刷新位置 应该是 commitedPosition <= flushedPosition
接下来，主要来看 MappedFileQueue comit flush 方法
3.3 MappedFileQueue #commit
public boolean commit(final int commitLeastPages) {
boolean result = true;
MappedFile mappedFile = this.findMappedFileByOffset(this.committedWhere, false); // @1
if (mappedFile != null) {
int offset = mappedFile.commit(commitLeastPages); // @2
long where = mappedFile.getFileFromOffset() + offset; // @3
result = where == this.committedWhere; // @4
this.committedWhere = where; // @5
} 

    return result;
}

代码@1：根据 committedWhere 找到具体的 MappedFile 文件
代码@2：调用 MappedFile 的 commit 函数
代码@3，mappedFile 返回的应该是当前 commit 的偏移量，加上该文件开始的偏移，，表示 MappedFileQueue 当前的提交偏移量
代码@4：如果 result = true,则可以认为 MappedFile#commit 本次并没有执行 commit 操作
代码@5，更新当前的 ccomitedWhere 指针。
接下来继续查看 MappedFile#commit 的实现：
MappedFile#commit()
public int commit(final int commitLeastPages) { // @1
if (writeBuffer == null) { // @2
//no need to commit data to file channel, so just regard wrotePosition as committedPosition.
return this.wrotePosition.get();
}
if (this.isAbleToCommit(commitLeastPages)) { // @3
if (this.hold()) {
commit0(commitLeastPages);
this.release();
} else {
log.warn(“in commit, hold failed, commit offset = ” + this.committedPosition.get());
}
}

    // All dirty data has been committed to FileChannel.
    if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
        this.transientStorePool.returnBuffer(writeBuffer);
        this.writeBuffer = null;
    }

    return this.committedPosition.get();
}

代码@1：参数，commitLeastPages 至少提交的页数，如果当前需要提交的数据所占的页数小于 commitLeastPages ，则不执行本次提交操作
代码@2：如果 writeBuffer 等于 null,则表示 IO 操作都是直接基于 FileChannel,所有，此时返回当前可写的位置，作为 committedPosition 即可，这里应该就有点 commit 是个啥意思了，
如果数据先写入到 writeBuffer 中，则需要提交到 FileChannel(MappedByteBuffer mappedByteBuffer)
代码@3：判断是否可以执行提交操作
protected boolean isAbleToCommit(final int commitLeastPages) {
int flush = this.committedPosition.get();
int write = this.wrotePosition.get();

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

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

    return write > flush;
}

从代码可以看出
@1 如果文件写满（this.fileSize == this.wrotePosition.get()) 则可以执行 commit
@2 如果有最小提交页数要求，则（当前写入位置/ pagesize(4k) – 当前 flush 位置/pagesize(4k) 大于 commitLeastPages 时，再提交。
@3，如果没有最新提交页数要求，则只有当前写入位置大于 flush，则可提交。
代码@4：执行具体的提交操作
protected void commit0(final int commitLeastPages) {
int writePos = this.wrotePosition.get();
int lastCommittedPosition = this.committedPosition.get();

    if (writePos - this.committedPosition.get() > 0) {
        try {
            ByteBuffer byteBuffer = writeBuffer.slice();    // @1
            byteBuffer.position(lastCommittedPosition);
            byteBuffer.limit(writePos);
            this.fileChannel.position(lastCommittedPosition);
            this.fileChannel.write(byteBuffer);      // @2
            this.committedPosition.set(writePos); // @3
        } catch (Throwable e) {
            log.error("Error occurred when commit data to FileChannel.", e);
        }
    }
}

代码@1，这里使用 slice 方法，主要是用的同一片内存空间，但单独的指针。
代码@2：将 bytebuf 当前 上一次 commitedPosition + 当前写位置这些数据全部写入到 FileChannel 中，commit 的左右原来是这要的，是将 writeBuffer 中的数据写入到 FileChannel 中
代码@3：更新 committedPosition 的位置。
讲到这里，commit 的作用就非常明白了，为了加深理解，该是来理解 MappedFile 几个核心属性的时候了。
protected int fileSize; // 文件的大小
protected FileChannel fileChannel; // 文件通道
/**

  • Message will put to here first, and then reput to FileChannel if writeBuffer is not null.
    */
    protected ByteBuffer writeBuffer = null; // 如果不为空，内容写写入到 writeBuffer,然后再重新放入到 FileChannel 中，这个重新放入，其实就是 commit 操作。
    protected TransientStorePool transientStorePool = null; // 临时存储，只有 pool 不为空，wrtieBuffer 才不会为空，也就是 MessageStoreConfig 中 transientStorePoolEnable 设置为 true 时
    // 才会生效，也是 writeBuffer 的容器。也就是 writeBuffer 其实就是堆内内存，如果 transientStorePoolEnable 为 true,消息是先直
    // 接放入到堆内存中，然后定时 commit 到堆外内存（FileChannel,MappedByteBuffer）中，再定时 flush.
    private long fileFromOffset; // 文件初始偏移量（在整个 MappedFile 链表中的逻辑偏移量）
    private File file; // 物理文件
    private MappedByteBuffer mappedByteBuffer; // mappedByteBufer,FileChanel 的内存映射。
    如果启用了 MessageStoreConfig 的 transientStorePoolEnable=true,消息在追加时，先放入到 writeBuffer 中，然后定时 commit 到 FileChannel,,然后定时 flush，如果 transientStorePoolEnable=false（默认）则消息追加时，直接存入 MappedByteBuffer 中，然后定时 flush ,【备注，说的是异步刷盘，如果是同步刷盘】，应该是直接调用 flush 方法。
    接下来我们再看一下 flush 方法，其实基本明了了，就是调用 FileChannel 的 force()方法。
    3.4 MappedFileQueue#flush MappedFile#flush
    public boolean flush(final int flushLeastPages) {
    boolean result = true;
    MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, false);
    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;
    }
    } return result; }
    /**
    • @return The current flushed position
      */
      public int flush(final int flushLeastPages) {
      if (this.isAbleToFlush(flushLeastPages)) {
      if (this.hold()) {
      int value = getReadPosition(); try { //We only append data to fileChannel or mappedByteBuffer, never both. if (writeBuffer != null || this.fileChannel.position() != 0) { this.fileChannel.force(false); } else { this.mappedByteBuffer.force(); } } 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();
      }
      具体代码很好理解，就不一一分析了。
      RocketMQ的刷盘机制就介绍到这，我们再简单做个总结
      先讲一下RocketMQ的存储设计亮点：（以 CommitLog 为例）
      单个 commitlog 文件，默认大小为 1G,由多个 commitlog 文件来存储所有的消息，commitlog 文件的命名以该文件在整个 commitlog 中的偏移量来命名，举例如下，
      例如一个 commitlog 文件，1024 个字节，
      第一个文件： 00000000000000000000
      第二个文件： 00000000000000001024
      MappedFile 封装一个一个的 CommitLog 文件，而 MappedFileQueue 就是封装的就是一个逻辑的 commitlog 文件。mappedFile 队列，从小到大排列。
      使用内存映射机制，MappedByteBuffer,具体封装类为 MappedFile。
      1、同步刷盘 每次发送消息，消息都直接存储在 FileChannel 中，使用的是(MapFile 的 mappdByteBuffer),然后直接调用 force()方法刷写到磁盘，等到 force 刷盘成功后，再返回给调用发（GroupCommitRequest#waitForFlush）就是其同步调用的实现。
      2、异步刷盘
      分为两种情况，是否开启堆内存缓存池，具体配置参数：MessageStoreConfig#transientStorePoolEnable
      transientStorePoolEnable=true
      消息在追加时，先放入到 writeBuffer 中，然后定时 commit 到 FileChannel,,然后定时 flush，
      transientStorePoolEnable=false（默认）
      消息追加时，直接存入 MappedByteBuffer 中，然后定时 flush ,【备注，说的是异步刷盘，如果是同步刷盘】，应该是直接调用 flush 方法。

MappedFile 重要的指针
wrotePosition：当前写入的指针
committedPosition : 上一次提交的指针 (transientStorePoolEnable=true 时有效)
flushedPosition ： 上一次 flush 的指针
OS_PAGE_SIZE = 1024 * 4 : 一页大小，4K

flushedPosition <= committedPosition <= wrotePosition <= fileSIze

作者：唯有坚持不懈
来源：CSDN
原文：https://blog.csdn.net/prestigeding/article/details/79188383
版权声明：本文为博主原创文章，转载请附上博文链接！

露水湾 , 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权
转载请注明原文链接：源码分析RocketMQ刷盘机制
喜欢 (0)赏
[]
分享 (0)
关于作者：
作者主页
发表我的评论
取消评论

表情 贴图 加粗 删除线 居中 斜体 签到

Hi，您需要填写昵称和邮箱！

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址