天天看点
返回

Rocket MQ : 拒绝神化零拷贝

注:本文绝非对零拷贝机制的否定 笔者能力有限，理解偏差请大家多多指正

不可否认零拷贝对于Rocket MQ的高性能表现有着积极正面的作用，但是笔者认为只是锦上添花，并非决定性因素。Rocket MQ性能卓越的原因绝非零拷贝就可以一言以蔽之。本文侧重点也不在这里，关于零拷贝已经有很多帖子

笔者企图从源码以及Linux内核背后探讨一下其他可能的原因。

预热机制

Rocket MQ采用内存映射来提高文件I/O访问性能，MappedFile、MappedFileQueue管理存储文件。MappedFileQueue对存储文件进行封装可以理解为MappedFile的管理容器。譬如CommitLog文件存储位置：${ROCKE_HOME}/store/commitlog/该目录下存在多个MappedFile文件。

MappedFile是内存映射的具体实现:构造方法包含文件名称、文件大小、以及一个transientStorePool标识位，如果开启transientStorePoolEnable机制则表示内容先存储在堆外内存，而后通过Commit线程将数据提交到FileChannel，而后Flush线程负责持久化。

public MappedFile(String fileName, int fileSize) throws IOException {
    init(fileName, fileSize);
}

public MappedFile(String fileName, int fileSize,
    TransientStorePool transientStorePool) throws IOException {
    init(fileName, fileSize, transientStorePool);
}
复制代码

两个构造方法不约而同的都会走init()，我们看两参数的那个即可。

private void init(String fileName, int fileSize) throws IOException {
    this.fileName = fileName;
    this.fileSize = fileSize;
    this.file = new File(fileName);
    this.fileFromOffset = Long.parseLong(this.file.getName());
    boolean ok = false;

    ensureDirOK(this.file.getParent());

    try {
        this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
        /**
         * fileChannel.map(MapMode mode, long position, long size)
         * 将此 fileChannel 对应的一个区域直接映射到内存中
         */
        this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
        /* 映射内存大小累加 */
        TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
        /* 映射文件个数累加 */
        TOTAL_MAPPED_FILES.incrementAndGet();
        ok = true;
    } catch (FileNotFoundException e) {
        log.error("Failed to create file " + this.fileName, e);
        throw e;
    } catch (IOException e) {
        log.error("Failed to map file " + this.fileName, e);
        throw e;
    } finally {
        if (!ok && this.fileChannel != null) {
            this.fileChannel.close();
        }
    }
}
复制代码

Java中在尝试文件映射的时候提供三种模式：

  • MapMode.READ_ONLY: 任何修改缓冲区的尝试将导致抛出ReadOnlyBufferException
  • MapMode.READ_WRITE:对结果缓冲区所做的更改将最终被传播到文件中
  • MapMode.PRIVATE: 对结果缓冲区所做的更改不会传播到文件中，其他程序不可见

值得注意的是映射一旦建立成功，就不再依赖fileChannel，即使此时关闭通道也不会影响映射的有效性，因此可以根据实际情况决定要不要close。

如果了解Linux内核的话，请您一定要注意直到此时为该文件分配的映射空间都是虚拟内存，并没有真的关联物理内存，当程序需要而物理内存又没有分配的时候则会触发一个Page Fault交由内核处理：

Rocket MQ : 拒绝神化零拷贝

上图展示的只是一个大概过程，实际情况复杂很多，因为缺页处理程序必须应对多种细分的特殊情况，(参见《深入理解LINUX内核》378页)，CommitLog文件大小固定为1G，如此大内存空间读写操作势必造成大量的缺页中断，显然这里绝对存在大量优化空间的。我们看看Rocket MQ作者如何优化。

不妨跟随笔者视角一探CommitLog如何获取MappedFile文件。

public MappedFile getLastMappedFile(long startOffset) {
    return getLastMappedFile(startOffset, true);
}
复制代码

getLastMappedFile方法会往AllocateMappedFileService#requestQueue阻塞队列提交AllocateRequest任务。AllocateMappedFileService服务线程此时会被唤醒执行mmapOperation方法。大致流程:

  • 阻塞队列requestQueue.take()出来一个任务对象，服务线程被唤醒，拿到AllocateRequest对象
  • 判断是否开启内存读写分离机制，决定选择如何构造MappedFile。 
/* 是否开启内存读写分离 */
if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
    try {
        /* Rocket允许自己定制实现细节 */
        mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
        mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
    } catch (RuntimeException e) {
        /* 没有自定义实现，使用系统默认实现 */
        log.warn("Use default implementation.");
        /* 注意这里三参构造 */
        mappedFile = new MappedFile(
            req.getFilePath(),
            req.getFileSize(), 
            messageStore.getTransientStorePool()
        );
    }
}
else {
    /* 注意这里两参构造 */
    mappedFile = new MappedFile(
        req.getFilePath(),
        req.getFileSize()
    );
}
复制代码
  • 源码这里将文件预热叫Pre write mappedFile，warmMappedFile方法负责具体的预热行为。这里这么做的原因是直接将缺页中断提前至初始化阶段，后续就不会因为频繁中断导致性能下降 
public void warmMappedFile(FlushDiskType type, int pages) {
    long beginTime = System.currentTimeMillis();
    ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
    int flush = 0;
    long time = System.currentTimeMillis();
    for (int i = 0, j = 0; i < this.fileSize; i += MappedFile.OS_PAGE_SIZE, j++) {
        byteBuffer.put(i, (byte) 0);
        /* force flush when flush disk type is sync */
        if (type == FlushDiskType.SYNC_FLUSH) {
            /* 每写入 pages 个内存页时刷盘一次 */
            if ((i / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE) >= pages) {
                flush = i;
                mappedByteBuffer.force();
            }
        }

        /* prevent gc */
        if (j % 1000 == 0) {
            log.info("j={}, costTime={}", j, System.currentTimeMillis() - time);
            time = System.currentTimeMillis();
            try {
                Thread.sleep(0);
            } catch (InterruptedException e) {
                log.error("Interrupted", e);
            }
        }
    }

    /* force flush when prepare load finished */
    if (type == FlushDiskType.SYNC_FLUSH) {
        log.info("mapped file warm-up done, force to disk, mappedFile={}, costTime={}",
            this.getFileName(), System.currentTimeMillis() - beginTime
        );
        mappedByteBuffer.force();
    }
    log.info("mapped file warm-up done. mappedFile={}, costTime={}",
        this.getFileName(), System.currentTimeMillis() - beginTime
    );
    
    /* !!! 这一行超级重要 !!! */
    this.mlock();
}
复制代码

mlock

  • 真香！Linux 原来是这么管理内存的
  • 一步一图带你深入理解 Linux 虚拟内存管理
  • 一步一图带你深入理解 Linux 物理内存管理

这里需要一点点Linux内核管理内存的前置知识：不了解的朋友可以稍微了解一下swap的概念。内存可以说是计算机系统中最为宝贵的资源了，再怎么多也不够用，当系统运行时间长了之后，难免会遇到内存紧张的时候，这时候就需要内核将那些不经常使用的内存页面回收起来，或者将那些可以迁移的页面进行内存规整，从而可以腾出连续的物理内存页面供内核分配。

简而言之就是当物理内存紧张的时候Linux内核会将别的进程的占用的物理内存swap到交换区(目前个人了解大部分都是磁盘)。

如此一来同一物理机如果有更加需要内存资源的进程，Linux内核完全有可能将我们通过预热机制好不容易全部都分配好的内存全部交换出去，这样Rocket MQ的性能一定呈现断崖式的下跌。

有没有一种机制使得进程可以独占一部分物理内存，不允许内核交换呢？神说要有光，于是Linux就暴露了mlock system call，而且Rocket MQ就是这么做的，上文提到的warmMappedFile方法的最后一行this.mlock就是用来lock memory的。

查阅一下手册就知道Linux提供了mlock, mlock2, munlock, mlockall, munlockall用来locK和unlock内存。

#include <sys/mman.h>

int mlock(const void *addr, size_t len);
int mlock2(const void *addr, size_t len, int flags);
int munlock(const void *addr, size_t len);

int mlockall(int flags);
int munlockall(void);
复制代码

总结一下就是Rocket MQ为了自身的高性能拒绝内存被操作系统交换

madvise

为了防止剧透，刚刚一直没有带大家看看MappedFile#mlock其实该方法还有别的妙处。

public void mlock() {
    long beginTime = System.currentTimeMillis();
    long address = ((DirectBuffer) (this.mappedByteBuffer)).address();
    Pointer pointer = new Pointer(address);
    {
        int ret = LibC.INSTANCE.mlock(pointer, new NativeLong(this.fileSize));
        log.info("mlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime);
    }

    {
        /**
         * MADV_WILLNEED 表示应用程序希望很快访问此地址范围
         */
        int ret = LibC.INSTANCE.madvise(pointer, new NativeLong(this.fileSize), LibC.MADV_WILLNEED);
        log.info("madvise {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime);
    }
}
复制代码

为了更加极致的性能体验，Linux操作系统暴露了madvise sysytem call ，madvise()系统调用，用于向内核提供对于起始地址为addr，长度为length的内存空间的操作建议或者指示。在大多数情况下，此类建议的目标是提高系统或者应用程序的性能。

#include <sys/mman.h>

int madvise(void *addr, size_t length, int advice);
复制代码

最初，此系统调用，仅仅支持一组常规的(conventional)建议值，这些建议值在各种系统中也有实现，(但是请注意，POSIX中并没有指定madvise())，后来又添加了许多特定于Linux的建议值。第三个参数advice其实就是一个标识，根据标识不同Linux内核采取的策略也有所区别。

  • Conventional advice values MADV_NORMAL：不做任何特殊处理，这是默认操作 MADV_RANDOM：期望以随机的顺序访问page，这等价于告诉内核，随机性强，局部性弱，预读机制意义不大 MADV_SEQUENTIAL：与MADV_RANDOM相反，期望顺序的访问page，因此内核应该积极的预读给定范围内的page，并在访问过后快速释放 MADV_WILLNEED：预计不久将会被访问，因此提前预读几页是个不错的主意 MADV_DONTNEED：与MADV_WILLNEED相反，预计未来长时间不会被访问，可以认为应用程序完成了对这部分内容的访问，因此内核可以释放与之相关的资源
  • Linux-specific advice values：Rocket MQ用的就是常规值，然后Linux特定值又特别多，所以这里挑选几个讲一下 MADV_DONTFORK:在执行fork(2)后，子进程不允许使用此范围的页面。这样是为了避免COW机制导致父进程在写入页面时更改页面的物理位置 MADV_DOFORK：撤销MADV_DONTFORK的效果，恢复默认行为 MADV_NOHUGEPAGE：确保指定范围内的页面不会使用透明大页。

Rocket MQ使用的是MADV_WILLNEED建议值，每次会预取提高性能。

文件系统设计

针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog文件。

Rocket MQ采用混合型存储结构，多个Topic的消息实体内容都存储于一个CommitLog(不包含因为文件写满，更换下一个文件的情况)，这使得所有的消息数据全部都是顺序写入该文件。然后RocketMQ使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue(逻辑消费队列)和IndexFile(索引文件)数据。虽然说因为还要落盘另外两种索引文件导致Rocket MQ其实没有办法保证全局的顺序写，但这两种文件其实足够小，况且索引文件自身也是顺序写，同时因为索引文件的特殊作用，也不可能将他们与数据文件相合并，可以说Rocket MQ已经尽最大努力保证全局顺序写了。

硬件加持

页缓存（PageCache)是OS对文件的缓存，用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。

在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在page cache机制的预读取作用下，Consume Queue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Deadline”（此时块存储采用SSD的话），随机读的性能也会有所提升。

另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存地址中，将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（正因为需要使用内存映射机制，故RocketMQ的文件存储都使用定长结构来存储，方便一次将整个文件映射至内存）。

上面提到的相邻文件预读、Mmap内存映射本质原因都是因为可以向内存借力，没有更强大的内存硬件一切都是空谈。其实软件工程师所能做的相对有限，我们只是在最大限度的发挥硬件的能力。

总结

笔者认为Rocket MQ高性能的关键是：

  • 内存加持，充分发挥硬件能力
  • 文件预热，将中断响应提前到初始化阶段
  • mlock禁止Linux交换内存
  • madvise向操作系统提出内存空间的操作建议或者指示
  • 优秀的文件系统设计，尽最大可能保证顺序写
  • 将磁盘上的物理文件直接映射到用户态的内存地址中(这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销)

引用

  • 《深入理解LINUX内核》
  • 《linux手册翻译——madvise(2)》
  • 《官方文档》
来源：https://juejin.cn/post/7169913280654213151
上一篇: RocketMQ源码分析之订阅消费组管理组件SubscriptionGroupManager
下一篇: WinForm应用实战开发指南 - 如何实现多语言开发框架模块的整合？