RocketMQ 整合 DLedger(多副本)即主从切换实现平滑升级的设计技巧

源码分析 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 构造方法

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

代码@1：如果开启 DLedger ，commitlog 的实现类为 DLedgerCommitLog，也是本文需要关注的关键所在。

代码@2：如果未开启 DLedger，则使用旧版的 Commitlog实现类。

2.2 增加节点状态变更事件监听器

BrokerController#initialize

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

主要调用 LedgerLeaderElector 的 addRoleChanneHandler 方法增加 节点角色变更事件监听器，DLedgerRoleChangeHandler 是实现主从切换的另外一个关键点。

2.3 调用 DefaultMessageStore 的 load 方法

DefaultMessageStore#load

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

代码@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 构造方法

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

代码@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

1public boolean load() {
2    boolean result = super.load();
3    if (!result) {
4        return false;
5    }
6    return true;
7}

DLedgerCommitLog 的 laod 方法实现比较简单，就是调用 其父类 Commitlog 的 load 方法，即这里也是为了启用 DLedger 时能够兼容以前的消息。

3.4 recover

在 Broker 启动时会加载 commitlog、consumequeue等文件，需要恢复其相关是数据结构，特别是与写入、刷盘、提交等指针，其具体调用 recover 方法。
DLedgerCommitLog#recover

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

首先会先恢复 consumequeue，得出 consumequeue 中记录的最大有效物理偏移量，然后根据该物理偏移量进行恢复。
接下来看一下该方法的处理流程与关键点。

DLedgerCommitLog#recover

1dLedgerFileStore.load();

Step1：加载 DLedger  相关的存储文件，并一一构建对应的 MmapFile，其初始化三个重要的指针 wrotePosition、flushedPosition、committedPosition 三个指针为文件的大小。

DLedgerCommitLog#recover

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

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

1isInrecoveringOldCommitlog = true;
2super.recoverNormally(maxPhyOffsetOfConsumeQueue);
3isInrecoveringOldCommitlog = false;

Step3：如果启用了 DLedger 并且是初次启动(还未生成 DLedger 相关的日志文件)，则需要恢复 旧的 commitlog 文件。

DLedgerCommitLog#recover

 1MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
 2if (mappedFile == null) {           // @1
 3    return;
 4}
 5ByteBuffer byteBuffer =  mappedFile.sliceByteBuffer();
 6byteBuffer.position(mappedFile.getWrotePosition());
 7boolean needWriteMagicCode = true;
 8// 1 TOTAL SIZE
 9byteBuffer.getInt(); //size
10int magicCode = byteBuffer.getInt();
11if (magicCode == CommitLog.BLANK_MAGIC_CODE) {   // @2
12    needWriteMagicCode = false;
13} else {
14    log.info("Recover old commitlog found a illegal magic code={}", magicCode);
15}
16dLedgerConfig.setEnableDiskForceClean(false);
17dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize();   // @3
18log.info("Recover old commitlog needWriteMagicCode={} pos={} file={} dividedCommitlogOffset={}", needWriteMagicCode, mappedFile.getFileFromOffset() + mappedFile.getWrotePosition(), mappedFile.getFileName(), dividedCommitlogOffset);
19if (needWriteMagicCode) {  // @4
20    byteBuffer.position(mappedFile.getWrotePosition());
21    byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition());
22    byteBuffer.putInt(BLANK_MAGIC_CODE);
23    mappedFile.flush(0);
24}
25mappedFile.setWrotePosition(mappedFile.getFileSize());   // @5
26mappedFile.setCommittedPosition(mappedFile.getFileSize());
27mappedFile.setFlushedPosition(mappedFile.getFileSize());
28dLedgerFileList.getLastMappedFile(dividedCommitlogOffset);
29log.info("Will set the initial commitlog offset={} for dledger", dividedCommitlogOffset);
30}

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技术内幕》一书。

DLedgerCommitLog#putMessage

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

关键点一：消息追加时，则不再写入到原先的 commitlog 文件中，而是调用 DLedgerServer 的 handleAppend 进行消息追加，该方法会有集群内的 Leader 节点负责消息追加以及在消息复制，只有超过集群内的半数节点成功写入消息后，才会返回写入成功。如果追加成功，将会返回本次追加成功后的起始偏移量，即 pos 属性，即类似于 rocketmq 中 commitlog 的偏移量，即物理偏移量。

DLedgerCommitLog#putMessage

1long wroteOffset =  dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
2ByteBuffer buffer = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
3String msgId = MessageDecoder.createMessageId(buffer, msg.getStoreHostBytes(), wroteOffset);
4eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
5appendResult = 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、从消息读取角度来看如何实现无缝兼容

---

DLedgerCommitLog#getMessage

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

消息查找比较简单，因为返回给客户端消息，转发给 consumequeue 的消息物理偏移量并不是  DLedger 条目的偏移量，而是真实消息的起始偏移量。其实现关键点如下：

• 如果查找的物理偏移量小于 dividedCommitlogOffset，则从原先的 commitlog 文件中查找。

• 然后根据物理偏移量按照二分方找到具体的物理文件。

• 对物理偏移量取模，得出在该物理文件中中的绝对偏移量，进行消息查找即可，因为只有知道其物理偏移量，从该处先将消息的长度读取出来，然后即可读出一条完整的消息。

6、总结

---

根据上面详细的介绍，我想读者朋友们应该不难得出如下结论：

• 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协议

---

更多文章请关注中间件兴趣圈公众号：