目錄

• 1、閱讀原始碼之前的思考
• 2、從 Broker 啟動流程看 DLedger
  • 2.1 構建 DefaultMessageStore
  • 2.2 增加節點狀態變更事件監聽器
  • 2.3 呼叫 DefaultMessageStore 的 load 方法
• 3、DLedgerCommitLog 詳解
  • 3.1 核心類圖
  • 3.2 構造方法
  • 3.3 load
  • 3.4 recover
• 4、從訊息追加看 DLedger 整合 RocketMQ 如何實現無縫相容
• 5、從訊息讀取看 DLedger 整合 RocketMQ 如何實現無縫相容
• 5、總結

原始碼分析 RocketMQ DLedger 多副本系列已經進行到第 8 篇了，前面的章節主要是介紹了基於 raft 協議的選主與日誌複製，從本篇開始將開始關注如何將 DLedger 應用到 RocketMQ中。

摘要：詳細分析了RocketMQ DLedger 多副本(主從切換) 是如何整合到 RocketMQ中，本文的行文思路首先結合已掌握的DLedger 多副本相關的知識初步思考其實現思路，然後從 Broker啟動流程、DLedgerCommitlog 核心類的講解，再從訊息傳送(追加)與訊息查詢來進一步探討 DLedger 是如何支援平滑升級的。

@(本節目錄)

1、閱讀原始碼之前的思考

RocketMQ 的訊息儲存檔案主要包括 commitlog 檔案、consumequeue 檔案與 Index 檔案。commitlog 檔案儲存全量的訊息，consumequeue、index 檔案都是基於 commitlog 檔案構建的。要使用 DLedger 來實現訊息儲存的一致性，應該關鍵是要實現 commitlog 檔案的一致性，即 DLedger 要整合的物件應該是 commitlog 檔案，即只需保證 raft 協議的複製組內各個節點的 commitlog 檔案一致即可。

我們知道使用檔案儲存訊息都會基於一定的儲存格式，rocketmq 的 commitlog 一個條目就包含魔數、訊息長度，訊息屬性、訊息體等，而我們再來回顧一下 DLedger 日誌的儲存格式：

DLedger 要整合 commitlog 檔案，是不是可以把 rocketmq 訊息，即一個個 commitlog 條目整體當成 DLedger 的 body 欄位即可。

還等什麼，跟我一起來看原始碼吧！！！別急，再拋一個問題，DLedger 整合 RocketMQ commitlog，能不能做到平滑升級？

帶著這些思考和問題，一起來探究 DLedger 是如何整合 RocketMQ 的。

2、從 Broker 啟動流程看 DLedger

溫馨提示：本文不會詳細介紹 Broker 端的啟動流程，只會點出在啟動過程中與 DLedger 相關的程式碼，如想詳細瞭解 Broker 的啟動流程，建議關注筆者的《RocketMQ技術內幕》一書。

Broker 涉及到 DLedger 相關關鍵點如下：

2.1 構建 DefaultMessageStore

DefaultMessageStore 構造方法

if(messageStoreConfig.isEnableDLegerCommitLog()) {  // @1
    this.commitLog = new DLedgerCommitLog(this);
 else {
    this.commitLog = new CommitLog(this);                    // @2
}

程式碼@1：如果開啟 DLedger ，commitlog 的實現類為 DLedgerCommitLog，也是本文需要關注的關鍵所在。

程式碼@2：如果未開啟 DLedger，則使用舊版的 Commitlog實現類。

2.2 增加節點狀態變更事件監聽器

BrokerController#initialize

if (messageStoreConfig.isEnableDLegerCommitLog()) {
    DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
    ((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
}

主要呼叫 LedgerLeaderElector 的 addRoleChanneHandler 方法增加 節點角色變更事件監聽器，DLedgerRoleChangeHandler 是實現主從切換的另外一個關鍵點。

2.3 呼叫 DefaultMessageStore 的 load 方法

DefaultMessageStore#load

// load Commit Log
result = result && this.commitLog.load();   // @1
// load Consume Queue
result = result && this.loadConsumeQueue();  
if (result) {
    this.storeCheckpoint =  new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
    this.indexService.load(lastExitOK);
    this.recover(lastExitOK);                         // @2
    log.info("load over, and the max phy offset = {}", this.getMaxPhyOffset());
}

程式碼@1、@2 最終都是委託 commitlog 物件來執行，這裡的關鍵又是如果開啟了 DLedger，則最終呼叫的是 DLedgerCommitLog。

經過上面的鋪墊，主角 DLedgerCommitLog “閃亮登場“了。

3、DLedgerCommitLog 詳解

溫馨提示：由於 Commitlog 的絕大部分方法都已經在《RocketMQ技術內幕》一書中詳細介紹了，並且 DLedgerCommitLog 的實現原理與 Commitlog 檔案的實現原理類同，本文會一筆帶過關於儲存部分的實現細節。

3.1 核心類圖

DLedgerCommitlog 繼承自 Commitlog。讓我們一一來看一下它的核心屬性。

• DLedgerServer dLedgerServer
  基於 raft 協議實現的叢集內的一個節點，用 DLedgerServer 例項表示。
• DLedgerConfig dLedgerConfig
  DLedger 的配置資訊。
• DLedgerMmapFileStore dLedgerFileStore
  DLedger 基於檔案對映的儲存實現。
• MmapFileList dLedgerFileList
  DLedger 所管理的儲存檔案集合，對比 RocketMQ 中的 MappedFileQueue。
• int id
  節點ID，0 表示主節點，非0表示從節點
• MessageSerializer messageSerializer
  訊息序列器。
• long beginTimeInDledgerLock = 0
  用於記錄 訊息追加的時耗(日誌追加所持有鎖時間)。
• long dividedCommitlogOffset = -1
  記錄的舊 commitlog 檔案中的最大偏移量，如果訪問的偏移量大於它，則訪問 dledger 管理的檔案。
• boolean isInrecoveringOldCommitlog = false
  是否正在恢復舊的 commitlog 檔案。

接下來我們將詳細介紹 DLedgerCommitlog 各個核心方法及其實現要點。

3.2 構造方法

public DLedgerCommitLog(final DefaultMessageStore defaultMessageStore) {
    super(defaultMessageStore);                   // @1
    dLedgerConfig =  new DLedgerConfig();
    dLedgerConfig.setEnableDiskForceClean(defaultMessageStore.getMessageStoreConfig().isCleanFileForciblyEnable());
    dLedgerConfig.setStoreType(DLedgerConfig.FILE);
    dLedgerConfig.setSelfId(defaultMessageStore.getMessageStoreConfig().getdLegerSelfId());
    dLedgerConfig.setGroup(defaultMessageStore.getMessageStoreConfig().getdLegerGroup());
    dLedgerConfig.setPeers(defaultMessageStore.getMessageStoreConfig().getdLegerPeers());
    dLedgerConfig.setStoreBaseDir(defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
    dLedgerConfig.setMappedFileSizeForEntryData(defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog());
    dLedgerConfig.setDeleteWhen(defaultMessageStore.getMessageStoreConfig().getDeleteWhen());
    dLedgerConfig.setFileReservedHours(defaultMessageStore.getMessageStoreConfig().getFileReservedTime() + 1);  
    id = Integer.valueOf(dLedgerConfig.getSelfId().substring(1)) + 1;            // @2
    dLedgerServer = new DLedgerServer(dLedgerConfig);                           // @3
    dLedgerFileStore = (DLedgerMmapFileStore) dLedgerServer.getdLedgerStore();
    DLedgerMmapFileStore.AppendHook appendHook = (entry, buffer, bodyOffset) -> {
            assert bodyOffset == DLedgerEntry.BODY_OFFSET;
            buffer.position(buffer.position() + bodyOffset + MessageDecoder.PHY_POS_POSITION);
            buffer.putLong(entry.getPos() + bodyOffset);
    };
    dLedgerFileStore.addAppendHook(appendHook);   // @4
    dLedgerFileList = dLedgerFileStore.getDataFileList();
    this.messageSerializer = new MessageSerializer(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());   // @5
}

程式碼@1：呼叫父類 即 CommitLog 的建構函式，載入 ${ROCKETMQ_HOME}/store/ comitlog 下的 commitlog 檔案，以便相容升級 DLedger 的訊息。我們稍微看一下 CommitLog 的建構函式：

程式碼@2：構建 DLedgerConfig 相關配置屬性，其主要屬性如下：

• enableDiskForceClean
  是否強制刪除檔案，取自 broker 配置屬性 cleanFileForciblyEnable，預設為 true 。
• storeType
  DLedger 儲存型別，固定為 基於檔案的儲存模式。
• dLegerSelfId
  leader 節點的 id 名稱，示例配置：n0，其配置要求第二個字元後必須是數字。
• dLegerGroup
  DLeger group 的名稱，建議與 broker 配置屬性 brokerName 保持一致。
• dLegerPeers
  DLeger Group 中所有的節點資訊，其配置示例 n0-127.0.0.1:40911;n1-127.0.0.1:40912;n2-127.0.0.1:40913。多個節點使用分號隔開。
• storeBaseDir
  設定 DLedger 的日誌檔案的根目錄，取自 borker 配件檔案中的 storePathRootDir ，即 RocketMQ 的資料儲存根路徑。
• mappedFileSizeForEntryData
  設定 DLedger 的單個日誌檔案的大小，取自 broker 配置檔案中的 - mapedFileSizeCommitLog，即與 commitlog 檔案的單個檔案大小一致。
• deleteWhen
  DLedger 日誌檔案的刪除時間，取自 broker 配置檔案中的 deleteWhen，預設為凌晨 4點。
• fileReservedHours
  DLedger 日誌檔案保留時長，取自 broker 配置檔案中的 fileReservedHours，預設為 72h。

程式碼@3：根據 DLedger 配置資訊建立 DLedgerServer，即建立 DLedger 叢集節點，叢集內各個節點啟動後，就會觸發選主。

程式碼@4：構建 appendHook 追加鉤子函式，這是相容 Commitlog 檔案很關鍵的一步，後面會詳細介紹其作用。

程式碼@5：構建訊息序列化。

根據上述的流程圖，構建好 DefaultMessageStore 實現後，就是呼叫其 load 方法，在啟用 DLedger 機制後，會依次呼叫 DLedgerCommitlog 的 load、recover 方法。

3.3 load

public boolean load() {
    boolean result = super.load();
    if (!result) {
        return false;
    }
    return true;
}

DLedgerCommitLog 的 laod 方法實現比較簡單，就是呼叫 其父類 Commitlog 的 load 方法，即這裡也是為了啟用 DLedger 時能夠相容以前的訊息。

3.4 recover

在 Broker 啟動時會載入 commitlog、consumequeue等檔案，需要恢復其相關是資料結構，特別是與寫入、刷盤、提交等指標，其具體呼叫 recover 方法。
DLedgerCommitLog#recover

public void recoverNormally(long maxPhyOffsetOfConsumeQueue) {  // @1
    recover(maxPhyOffsetOfConsumeQueue);
}

首先會先恢復 consumequeue，得出 consumequeue 中記錄的最大有效物理偏移量，然後根據該物理偏移量進行恢復。
接下來看一下該方法的處理流程與關鍵點。

DLedgerCommitLog#recover

dLedgerFileStore.load();

Step1：載入 DLedger 相關的儲存檔案，並一一構建對應的 MmapFile，其初始化三個重要的指標 wrotePosition、flushedPosition、committedPosition 三個指標為檔案的大小。

DLedgerCommitLog#recover

if (dLedgerFileList.getMappedFiles().size() > 0) {   
    dLedgerFileStore.recover();   // @1
    dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset();     // @2
    MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
    if (mappedFile != null) {                                                                                                       // @3
        disableDeleteDledger();
    }
    long maxPhyOffset = dLedgerFileList.getMaxWrotePosition();
    // Clear ConsumeQueue redundant data
    if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) {      // @4
        log.warn("[TruncateCQ]maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset);
        this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset);
    }
    return;
}

Step2：如果已存在 DLedger 的資料檔案，則只需要恢復 DLedger 相關資料文建，因為在載入舊的 commitlog 檔案時已經將其重要的資料指標設定為最大值。其關鍵實現點如下：

• 首先呼叫 DLedger 檔案儲存實現類 DLedgerFileStore 的 recover 方法，恢復管轄的 MMapFile 物件(一個檔案對應一個MMapFile例項)的相關指標，其實現方法與 RocketMQ 的 DefaultMessageStore 的恢復過程類似。
• 設定 dividedCommitlogOffset 的值為 DLedger 中所有物理檔案的最小偏移量。操作訊息的物理偏移量小於該值，則從 commitlog 檔案中查詢；物理偏移量大於等於該值的話則從 DLedger 相關的檔案中查詢訊息。
• 如果存在舊的 commitlog 檔案，則禁止刪除 DLedger 檔案，其具體做法就是禁止強制刪除檔案，並將檔案的有效儲存時間設定為 10 年。
• 如果 consumequeue 中儲存的最大物理偏移量大於 DLedger 中最大的物理偏移量，則刪除多餘的 consumequeue 檔案。

溫馨提示：為什麼當存在 commitlog 檔案的情況下，不能刪除 DLedger 相關的日誌檔案呢？

因為在此種情況下，如果 DLedger 中的物理檔案有刪除，則物理偏移量會斷層。

正常情況下， maxCommitlogPhyOffset 與 dividedCommitlogOffset 是連續的，這樣非常方便是訪問 commitlog 還是 訪問 DLedger ，但如果DLedger 部分檔案刪除後，這兩個值就變的不連續，就會造成中間的檔案空洞，無法被連續訪問。

DLedgerCommitLog#recover

isInrecoveringOldCommitlog = true;
super.recoverNormally(maxPhyOffsetOfConsumeQueue);
isInrecoveringOldCommitlog = false;

Step3：如果啟用了 DLedger 並且是初次啟動(還未生成 DLedger 相關的日誌檔案)，則需要恢復 舊的 commitlog 檔案。

DLedgerCommitLog#recover

MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
if (mappedFile == null) {           // @1
    return;
}
ByteBuffer byteBuffer =  mappedFile.sliceByteBuffer();
byteBuffer.position(mappedFile.getWrotePosition());
boolean needWriteMagicCode = true;
// 1 TOTAL SIZE
byteBuffer.getInt(); //size
int magicCode = byteBuffer.getInt();
if (magicCode == CommitLog.BLANK_MAGIC_CODE) {   // @2
    needWriteMagicCode = false;
} else {
    log.info("Recover old commitlog found a illegal magic code={}", magicCode);
}
dLedgerConfig.setEnableDiskForceClean(false);
dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize();   // @3
log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset);
if (needWriteMagicCode) {  // @4
    byteBuffer.position(mappedFile.getWrotePosition());
    byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition());
    byteBuffer.putInt(BLANK_MAGIC_CODE);
    mappedFile.flush(0);
}
mappedFile.setWrotePosition(mappedFile.getFileSize());   // @5
mappedFile.setCommittedPosition(mappedFile.getFileSize());
mappedFile.setFlushedPosition(mappedFile.getFileSize());
dLedgerFileList.getLastMappedFile(dividedCommitlogOffset);
log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset);
}

Step4：如果存在舊的 commitlog 檔案，需要將最後的檔案剩餘部分全部填充，即不再接受新的資料寫入，新的資料全部寫入到 DLedger 的資料檔案中。其關鍵實現點如下：

• 嘗試查詢最後一個 commitlog 檔案，如果未找到，則結束。
• 從最後一個檔案的最後寫入點(原 commitlog 檔案的 待寫入位點)嘗試去查詢寫入的魔數，如果存在魔數並等於 CommitLog.BLANK_MAGIC_CODE，則無需再寫入魔數，在升級 DLedger 第一次啟動時，魔數為空，故需要寫入魔數。
• 初始化 dividedCommitlogOffset ，等於最後一個檔案的起始偏移量加上檔案的大小，即該指標指向最後一個檔案的結束位置。
• 將最後一個 commitlog 未寫滿的資料全部寫入，其方法為 設定訊息體的 size 與 魔數即可。
• 設定最後一個檔案的 wrotePosition、flushedPosition、committedPosition 為檔案的大小，同樣有意味者最後一個檔案已經寫滿，下一條訊息將寫入 DLedger 中。

在啟用 DLedger 機制時 Broker 的啟動流程就介紹到這裡了，相信大家已經瞭解 DLedger 在整合 RocketMQ 上做的努力，接下來我們從訊息追加、訊息讀取兩個方面再來探討 DLedger 是如何無縫整合 RocketMQ 的，實現平滑升級的。

4、從訊息追加看 DLedger 整合 RocketMQ 如何實現無縫相容

溫馨提示：本節同樣也不會詳細介紹整個訊息追加(儲存流程)，只是要點出與 DLedger(多副本、主從切換)相關的核心關鍵點。如果想詳細瞭解訊息追加的流程，可以閱讀筆者所著的《RocketMQ技術內幕》一書。

DLedgerCommitLog#putMessage

AppendEntryRequest request = new AppendEntryRequest();
request.setGroup(dLedgerConfig.getGroup());
request.setRemoteId(dLedgerServer.getMemberState().getSelfId());
request.setBody(encodeResult.data);
dledgerFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
if (dledgerFuture.getPos() == -1) {
    return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR));
}

關鍵點一：訊息追加時，則不再寫入到原先的 commitlog 檔案中，而是呼叫 DLedgerServer 的 handleAppend 進行訊息追加，該方法會有叢集內的 Leader 節點負責訊息追加以及在訊息複製，只有超過叢集內的半數節點成功寫入訊息後，才會返回寫入成功。如果追加成功，將會返回本次追加成功後的起始偏移量，即 pos 屬性，即類似於 rocketmq 中 commitlog 的偏移量，即物理偏移量。

DLedgerCommitLog#putMessage

long wroteOffset =  dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
ByteBuffer buffer = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
String msgId = MessageDecoder.createMessageId(buffer, msg.getStoreHostBytes(), wroteOffset);
eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, encodeResult.data.length, msgId, System.currentTimeMillis(), queueOffset, eclipseTimeInLock);

關鍵點二：根據 DLedger 的起始偏移量計算真正的訊息的物理偏移量，從開頭部分得知，DLedger 自身有其儲存協議，其 body 欄位儲存真實的訊息，即 commitlog 條目的儲存結構，返回給客戶端的訊息偏移量為 body 欄位的開始偏移量，即通過 putMessage 返回的物理偏移量與不使用Dledger 方式返回的物理偏移量的含義是一樣的，即從開偏移量開始，可以正確讀取訊息，這樣 DLedger 完美的相容了 RocketMQ Commitlog。關於 pos 以及 wroteOffset 的圖解如下：

5、從訊息讀取看 DLedger 整合 RocketMQ 如何實現無縫相容

DLedgerCommitLog#getMessage

public SelectMappedBufferResult getMessage(final long offset, final int size) {
    if (offset < dividedCommitlogOffset) {   // @1
        return super.getMessage(offset, size);
    }
    int mappedFileSize = this.dLedgerServer.getdLedgerConfig().getMappedFileSizeForEntryData();
    MmapFile mappedFile = this.dLedgerFileList.findMappedFileByOffset(offset, offset == 0);   // @2
    if (mappedFile != null) {
        int pos = (int) (offset % mappedFileSize);
        return  convertSbr(mappedFile.selectMappedBuffer(pos, size));                                       // @3
    }
    return null;
}

訊息查詢比較簡單，因為返回給客戶端訊息，轉發給 consumequeue 的訊息物理偏移量並不是 DLedger 條目的偏移量，而是真實訊息的起始偏移量。其實現關鍵點如下：

• 如果查詢的物理偏移量小於 dividedCommitlogOffset，則從原先的 commitlog 檔案中查詢。
• 然後根據物理偏移量按照二分方找到具體的物理檔案。
• 對物理偏移量取模，得出在該物理檔案中中的絕對偏移量，進行訊息查詢即可，因為只有知道其物理偏移量，從該處先將訊息的長度讀取出來，然後即可讀出一條完整的訊息。

5、總結

根據上面詳細的介紹，我想讀者朋友們應該不難得出如下結論：

• DLedger 在整合時，使用 DLedger 條目包裹 RocketMQ 中的 commitlog 條目，即在 DLedger 條目的 body 欄位來儲存整條 commitlog 條目。
• 引入 dividedCommitlogOffset 變數，表示物理偏移量小於該值的訊息存在於舊的 commitlog 檔案中，實現 升級 DLedger 集群后能訪問到舊的資料。
• 新 DLedger 叢集啟動後，會將最後一個 commitlog 填充，即新的資料不會再寫入到 原先的 commitlog 檔案。
• 訊息追加到 DLedger 資料日誌檔案中，返回的偏移量不是 DLedger 條目的起始偏移量，而是DLedger 條目中 body 欄位的起始偏移量，即真實訊息的起始偏移量，保證訊息物理偏移量的語義與 RocketMQ Commitlog一樣。

RocketMQ 整合 DLedger(多副本)實現平滑升級的設計技巧就介紹到這裡了。

如果本文對您有一定的幫助話，麻煩幫忙點個贊，非常感謝。

---

推薦閱讀：原始碼分析 RocketMQ DLedger 多副本即主從切換系列文章：
1、RocketMQ 多副本前置篇：初探raft協議
2、原始碼分析 RocketMQ DLedger 多副本之 Leader 選主
3、原始碼分析 RocketMQ DLedger 多副本儲存實現
4、原始碼分析 RocketMQ DLedger(多副本) 之日誌追加流程
5、原始碼分析 RocketMQ DLedger(多副本) 之日誌複製(傳播)
6、基於 raft 協議的 RocketMQ DLedger 多副本日誌複製設計原理

---

作者介紹：丁威，《RocketMQ技術內幕》作者，RocketMQ 社群佈道師，公眾號：中介軟體興趣圈 維護者，目前已陸續發表原始碼分析Java集合、Java 併發包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等原始碼專欄。可以點選連結加入中介軟體知識星球 ，一起探討高併發、分散式服務架構，交流原始碼。

相關推薦