天天看點
傳回

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應用實戰開發指南 - 如何實作多語言開發架構子產品的整合？