2. 程式人生 > >RocketMQ原始碼分析----刷盤的實現

RocketMQ原始碼分析----刷盤的實現

阿新 • • 發佈：2019-01-22

注：可以先了解一下記憶體對映，然後再看RocketMq的刷盤，會更容易理解

Broker啟動的時候，會呼叫CommitLog的start方法，然後再啟動flushCommitLogService執行緒

在CommitLog的構造方法中，會判斷刷盤的型別

    public CommitLog(final DefaultMessageStore defaultMessageStore) {
        this.mapedFileQueue = new MapedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),
                defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog(), defaultMessageStore.getAllocateMapedFileService());
        this.defaultMessageStore = defaultMessageStore;
        
        if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
            this.flushCommitLogService = new GroupCommitService();
        } else {
            this.flushCommitLogService = new FlushRealTimeService();
        }

        this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());
    }
同步刷盤使用GroupCommitService，非同步刷盤使用FlushRealTimeService，預設是使用非同步刷盤

同步和非同步的區別在於，broker的處理Producer請求的時候，如果是同步刷盤，那麼會進行刷盤後才返回給Producer傳送成功，而非同步刷盤則是喚醒刷盤執行緒後就返回

非同步刷盤

非同步刷盤是FlushRealTimeService，其run方法有個while迴圈，只要broker不關閉就一直迴圈下去

public void run() {
    while (!this.isStoped()) {
        boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();//是否定時刷盤，預設為實時刷盤
//刷盤間隔，預設1秒
        int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
        int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();//刷CommitLog，至少刷幾個PAGE
//刷CommitLog，徹底刷盤間隔時間
        int flushPhysicQueueThoroughInterval =
                CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();

        boolean printFlushProgress = false;

        // Print flush progress
        long currentTimeMillis = System.currentTimeMillis();
        if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {//每flushPhysicQueueThoroughInterva秒徹底刷盤一次
 


            this.lastFlushTimestamp = currentTimeMillis;
            flushPhysicQueueLeastPages = 0;//這個引數為0後面會說到
            printFlushProgress = ((printTimes++ % 10) == 0);
        }

        try {
            if (flushCommitLogTimed) {//定時刷
                Thread.sleep(interval);
            } else {//實時刷
                this.waitForRunning(interval);
            }

            if (printFlushProgress) {
                this.printFlushProgress();//空方法
            }

            CommitLog.this.mapedFileQueue.commit(flushPhysicQueueLeastPages);
            long storeTimestamp = CommitLog.this.mapedFileQueue.getStoreTimestamp();
            if (storeTimestamp > 0) {
                CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
            }
        } catch (Exception 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 < RetryTimesOver && !result; i++) {
        result = CommitLog.this.mapedFileQueue.commit(0);
        CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
    }

}

首先獲取配置：刷盤間隔，刷盤時至少刷幾個page，徹底刷盤間隔時間等，然後真正刷盤的地方是CommitLog.this.mapedFileQueue.commit(flushPhysicQueueLeastPages);

    public boolean commit(final int flushLeastPages) {
        boolean result = true;
        MapedFile mapedFile = this.findMapedFileByOffset(this.committedWhere, true);
        if (mapedFile != null) {
            long tmpTimeStamp = mapedFile.getStoreTimestamp();
            int offset = mapedFile.commit(flushLeastPages);
            long where = mapedFile.getFileFromOffset() + offset;
            result = (where == this.committedWhere);
            this.committedWhere = where;
            if (0 == flushLeastPages) {
                this.storeTimestamp = tmpTimeStamp;
            }
        }

        return result;
    }

this.findMapedFileByOffset(this.committedWhere, true);是根據對應位移獲取對應的檔案

然後呼叫commit方法進行刷盤

    public int commit(final int flushLeastPages) {
        if (this.isAbleToFlush(flushLeastPages)) {
            if (this.hold()) {
                int value = this.wrotePostion.get();
                this.mappedByteBuffer.force();
                this.committedPosition.set(value);
                this.release();
            } else {
                log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
                this.committedPosition.set(this.wrotePostion.get());
            }
        }

        return this.getCommittedPosition();
    }

然後還要呼叫isAbleToFlush方法判斷是否可以刷盤，真正刷盤是ByteBuffer的force方法

    private boolean isAbleToFlush(final int flushLeastPages) {
        int flush = this.committedPosition.get();//當前刷盤刷到的位置
        int write = this.wrotePostion.get();//當前檔案寫到的位置

        // 如果當前檔案已經寫滿，應該立刻刷盤
        if (this.isFull()) {
            return true;
        }

        // 只有未刷盤資料滿足指定page數目才刷盤
        if (flushLeastPages > 0) {
            return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
        }

        return write > flush;
    }

第二個if的地方，如果當前寫的位置和上一次刷盤的位置之間相差是flushLeastPages個頁以上，才可以進行刷盤，所以即使是實時的刷盤也是要到達指定的頁數後才會進行刷盤

當flushLeastPages等於0的情況，只要寫的位置比上次刷盤的位置大就行了，而flushLeastPages是在FlushRealTimeService的run方法中設定的

        if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {//每flushPhysicQueueThoroughInterva秒徹底刷盤一次

            this.lastFlushTimestamp = currentTimeMillis;
            flushPhysicQueueLeastPages = 0;
            printFlushProgress = ((printTimes++ % 10) == 0);
        }
當到達徹底刷盤的時間後，就講flushLeastPages設定為0

當這個引數為0的時候，那麼只要有資料就進行刷盤操作

當這個方法返回true的時候就將ByteBuffer裡的資料刷到檔案中

綜上：broker啟動的時候就會啟動一個執行緒，去持續的進行刷盤操作


在處理broker傳送請求的時候，有一個判斷

// Synchronization flush
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
    GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
    if (msg.isWaitStoreMsgOK()) {//是否等待伺服器將訊息儲存完畢再返回
        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: " + msg.getTopic() + " tags: " + msg.getTags()
                    + " client address: " + msg.getBornHostString());
            putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
        }
    } else {
        service.wakeup();
    }
}
// Asynchronous flush
else {
    this.flushCommitLogService.wakeup();
}

wakeup方法如下

public void wakeup() {
    synchronized (this) {
        if (!this.hasNotified) {
            this.hasNotified = true;
            this.notify();
        }
    }
}

protected void waitForRunning(long interval) {
    synchronized (this) {
        if (this.hasNotified) {
            this.hasNotified = false;
            this.onWaitEnd();
            return;
        }

        try {
            this.wait(interval);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            this.hasNotified = false;
            this.onWaitEnd();
        }
    }
}

將標誌位設定為true，然後run的迴圈中，有行程式碼this.waitForRunning(interval);就是等待一段時間，然後再去刷盤，而呼叫了wakeup方法，就不用等待，直接就返回刷盤

另外，run方法的迴圈外有一段程式碼

// Normal shutdown, to ensure that all the flush before exit
boolean result = false;
for (int i = 0; i < RetryTimesOver && !result; i++) {
    result = CommitLog.this.mapedFileQueue.commit(0);
    CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
}

當出了迴圈，證明broker已經停止了，那麼需要將所有的資料進行刷盤。而且commit傳入的引數為0，那麼在判斷是否刷盤的時候只要寫的位置比上次刷盤的位置大就行了

總結：非同步刷盤情況下，broker會開啟一個執行緒，等待一段時間，然後判斷刷盤條件是否符合，若符合就進行刷盤。在迴圈中會重複這個操作。在傳送訊息的時候只會進行一個喚醒操作，然後等待的執行緒馬上返回進行判斷刷盤。

在broker停止只會還會將所有的資料進行刷盤



同步刷盤

非同步刷盤使用的是FlushRealTimeService，而同步刷盤使用的是GroupCommitService

然後傳送訊息的時候，有個判斷

// Synchronization flush
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
    GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
    if (msg.isWaitStoreMsgOK()) {//是否等待伺服器將訊息儲存完畢再返回
        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: " + msg.getTopic() + " tags: " + msg.getTags()
                    + " client address: " + msg.getBornHostString());
            putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
        }
    } else {
        service.wakeup();
    }
}

首先建立一個GroupCommitRequest物件，引數為刷盤後文件的位置，即當前寫到的位置+要刷盤的位元組數

接著呼叫putRequest方法

public void putRequest(final GroupCommitRequest request) {
    synchronized (this) {
        this.requestsWrite.add(request);
        if (!this.hasNotified) {
            this.hasNotified = true;
            this.notify();
        }
    }
}

把request放到list中，然後喚醒執行緒(這裡和非同步刷盤一樣)

request.waitForFlush方法使用CountDownLatch來阻塞當前執行緒，直到超時或者刷盤完成(GroupCommitService中呼叫countDown()方法)

和非同步刷盤一樣，一個while迴圈中執行刷盤操作，當broker正常停止會把為刷盤的資料進行刷盤，如下

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

    while (!this.isStoped()) {
        try {
            this.waitForRunning(0);
            this.doCommit();
        } catch (Exception e) {
            CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
        }
    }

    // Under normal circumstances shutdown, wait for the arrival of the
    // request, and then flush
    try {
        Thread.sleep(10);
    } catch (InterruptedException e) {
        CommitLog.log.warn("GroupCommitService Exception, ", e);
    }

    synchronized (this) {
        this.swapRequests();
    }

    this.doCommit();

    CommitLog.log.info(this.getServiceName() + " service end");
}

刷盤的真正邏輯在CommitLog類中的doCommit方法

private void doCommit() {
    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.mapedFileQueue.getCommittedWhere() >= req.getNextOffset());//上一次刷盤的位置是否小於這個刷盤後的位置

                if (!flushOK) {
                    CommitLog.this.mapedFileQueue.commit(0);//刷盤
                }
            }
            //CountDownLatch計數器減1，putMessage中的request.waitForFlush繼續執行
            req.wakeupCustomer(flushOK);
        }

        long storeTimestamp = CommitLog.this.mapedFileQueue.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.mapedFileQueue.commit(0);
    }
}

邏輯相對簡單，先判斷位置是否正確，即刷盤後的位置要大於上次刷盤後的位置，然後和非同步刷盤一樣，使用commit方法進行刷盤，看了非同步刷盤，就知道了引數為0，則只判斷位置，不管“至少要刷多少個頁”的這個配置，即有多少刷多少

刷完之後，會喚醒剛剛阻塞的執行緒

總結：

傳送訊息的時候，組裝GroupCommitRequest物件(儲存了刷盤位置資訊)，喚醒同步刷盤執行緒GroupCommitService，然後使用CountDownLatch阻塞當前執行緒直至超時或刷盤完成，然後正在的刷盤操作在CommitLog的doCommit方法中

相關推薦

RocketMQ原始碼分析----刷盤的實現

注：可以先了解一下記憶體對映，然後再看RocketMq的刷盤，會更容易理解 Broker啟動的時候，會呼叫CommitLog的start方法，然後再啟動flushCommitLogService執行緒 在CommitLog的構造方法中，會判斷刷盤的型別 publ

RocketMQ原始碼分析之RocketMQ事務訊息實現原理上篇(二階段提交)

根據上文的描述，傳送事務訊息的入口為： TransactionMQProducer#sendMessageInTransaction： public TransactionSendResult sendMessageInTransaction(final Message msg, final Object

RocketMQ原始碼分析之從官方示例窺探：RocketMQ事務訊息實現基本思想

RocketMQ4.3.0版本開始支援事務訊息，後續分享將開始將剖析事務訊息的實現原理。首先從官方給出的Demo例項入手，以此通往RocketMQ事務訊息的世界中。 官方版本未釋出之前，從apache rocketmq第一個版本上線後，程式碼中存在與事務訊息相關的程式碼，例如COMMIT、ROLLBACK、

【Android】原始碼分析 - LRUCache快取實現原理

一、Android中的快取策略 一般來說，快取策略主要包含快取的新增、獲取和刪除這三類操作。如何新增和獲取快取這個比較好理解，那麼為什麼還要刪除快取呢？這是因為不管是記憶體快取還是硬碟快取，它們的快取大小都是有限的。當快取滿了之後，再想其新增快取，這個時候就需要刪除一些舊的快取

OpenCV兩種畸變校正模型原始碼分析以及CUDA實現

http://www.cnblogs.com/riddick/p/7811877.html 影象演算法中會經常用到攝像機的畸變校正，有必要總結分析OpenCV中畸變校正方法，其中包括普通針孔相機模型和魚眼相機模型fisheye兩種畸變校正方法。 普通相機模型畸變校正函式針對OpenCV中的c

RocketMQ原始碼分析(一)之除錯環境搭建

版本宣告 基於rocketmq-all-4.3.1版本 所有分析是根據自己的理解，因為不是RocketMQ的原始開發者，所以肯定會存在分析不正確的地方，如有發現歡迎指正，謝謝！ 規定$ROCKET

[Vue原始碼分析] v-model實現原理

最近小組有個關於vue原始碼分析的分享會，提前準備一下… 前言: 我們都知道使用v-model可以實現資料的雙向繫結，及實現資料的變化驅動dom的更新，dom的更新影響資料的變化。那麼v-model是怎麼實現這一原理的呢？接下來探索一下這部分的原始碼。 前期準備 ①：vue2

RocketMQ 原始碼分析 訊息儲存（預備知識二）（轉載+整理）

前言 在RMQ中為了提高commitlog檔案的讀寫效率，而採用了一個叫做記憶體對映的技術。按照我的理解，記憶體對映在處理大檔案上有非常大的效能提升，所以這篇來記錄一下我對記憶體對映的理解。 使用者態和核心態 我們都知道作業系統分為使用者態和核心態，核心態表示當前為核心程式

RocketMQ 原始碼分析 訊息儲存（預備知識一）

前言 看到RocketMQ的效能問題的時候，通常能看到page cache、順序IO寫、預讀等，要想設計出一個高效能的中介軟體，這部分的知識是絕對要掌握的。 順序IO讀寫為什麼速度更快 當需要從硬碟上讀取一個檔案時，首先會要求磁頭定位到這個檔案的起始扇區。這個定位過程包

RocketMQ 原始碼分析 BrokerStartUp（七）

前言 broker是訊息接收處理，訊息落地的核心模組。這個模組用於接收producer傳送的訊息以及consumer消費的訊息。 本章從broker的啟動類入手分析一下broker啟動的過程。 分析broker是為了能夠搞明白master和slave之間的關係，以及訊息落地的具體

RocketMQ 原始碼分析 NettyRemotingServer（六）

前言 已經寫了幾篇關於RocketMQ原始碼的分析，可能對其邏輯處理會多一點認識，但還沒深入到RocketMQ精髓中。比如MQ中的通訊是如何實現的、如何實現高效能，高可用、最終一致性、MQ 訊息儲存。這些才是我們閱讀原始碼的一個目標。所以這篇通過分析原始碼的來了解一下RocketMQ

RocketMQ 原始碼分析 BrokerHousekeepingService（五）

前言 RocketMQ在早期版本使用的是Zookeeper，考慮到Zookeeper不夠輕量，後面用namesrv來代替。熟悉Zookeeper的知道它使用心跳檢測來判斷服務是否正常，那麼namesrv肯定也少不了這塊功能，那麼它是怎麼判斷一個Broker節點是否Live呢？如果一個

RocketMQ原始碼分析之順序消費

概述 RocketMQ按照順序消費有兩種順序級別，一種是普通順序訊息。另外一種是更完全嚴格順序 普通順序訊息指的是Producer將訊息傳送到相對應的訊息佇列上面 完全嚴格順序：在普通順序訊息的基礎上，Consumer嚴格進行順序消費 在絕大部分情況下只

Erlang：RabbitMQ原始碼分析 4. file_handle_cache實現分析

RabbitMQ的檔案操作使用file_handle_cache，將Erlang的prim_file Module包了一層。增加了writeBuffer和對檔案開啟數量的控制邏輯 file_handle_cache也是一個gen_server2，但一般來說只有Open操作

LinkedHashMap原始碼分析與LRU實現

LinkedHashMap可認為是雜湊表和連結列表綜合實現，並允許使用null值和null鍵。LinkedHashMap實現與HashMap的不同之處在於，LinkedHashMap維護著一個運行於所有條目的雙重連結列表。此連結列表定義了迭代順序，該迭代順序可以是插入

分散式訊息佇列RocketMQ原始碼分析之3 -- Consumer負載均衡機制 -- Rebalance

同Kafka一樣，RocketMQ也需要探討一個問題：如何把一個topic的多個queue分攤給不同的consumer，也就是負載均衡問題。 有興趣朋友可以關注公眾號“架構之道與術”， 獲取最新文章。 或掃描如下二維碼： 在討論這個問題之前，我們先看一

分散式訊息佇列RocketMQ原始碼分析之2 -- Broker與NameServer心跳機制

我們知道，Kafka是通過ZK的臨時節點來監測Broker的死亡的。當一個Broker掛了之後，ZK上面對應的臨時節點被刪除，同時其他Broker收到通知。 那麼在RocketMQ中，對應的NameServer是如何判斷一個Broker的死亡呢？ 有興趣朋友

RocketMQ原始碼分析之rocketmq-broker啟動 (二)

本文章會從以下幾個方面介紹rocketmq-broker啟動流程 1. Broker簡單介紹 1.1 initialize 1.2 start 2. broker啟動程式碼分析 2.1 r

分散式訊息佇列 RocketMQ 原始碼分析 —— Message 順序傳送與消費

本文主要基於 RocketMQ 4.0.x 正式版 1. 概述 建議前置閱讀內容： 當然對 Message 傳送與消費已經有一定了解的同學，可以選擇跳過。 RocketMQ 提供了兩種順序級別： 普通順序訊息 ：Producer 將相關聯的訊息傳送到相同

SprignMVC+myBatis整合+mybatis原始碼分析+動態代理實現流程+如何根據mapper介面生成其實現類

首先熟悉三個概念： SqlSessionFactoryBean –為整合應用提供SqlSession物件資源 MapperFactoryBean –根據指定的Mapper介面生成Bean例項 MapperScannerConfigurer –根據指定