首页 > 编程语言> > 源码分析 RocketMQ DLedger(多副本) 之日志追加流程

源码分析 RocketMQ DLedger(多副本) 之日志追加流程

2020-12-03 14:53:21 作者：互联网

上一篇我们详细分析了源码分析RocketMQ多副本之Leader选主，本文将详细分析日志复制的实现。

有了前篇源码分析 RocketMQ DLedger 多副本存储实现，本文将直接从 Leader 处理客户端请求入口开始，其入口为：DLedgerServer 的 handleAppend 方法开始讲起。

1、日志复制基本流程

在正式分析 RocketMQ DLedger 多副本复制之前，我们首先来了解客户端发送日志的请求协议字段，其类图如下所示：

源码分析 RocketMQ DLedger(多副本) 之日志追加流程
我们先一一介绍各个字段的含义：

String group
该集群所属组名。
String remoteId
请求目的节点ID。
String localId
节点ID。
int code
请求响应字段，表示返回响应码。
String leaderId = null
集群中的Leader Id。
long term
集群当前的选举轮次。
byte[] body
待发送的数据。
日志的请求处理处理入口为 DLedgerServer 的 handleAppend 方法。

DLedgerServer#handleAppend

1PreConditions.check(memberState.getSelfId().equals(request.getRemoteId()), DLedgerResponseCode.UNKNOWN_MEMBER, "%s != %s", request.getRemoteId(), memberState.getSelfId());
2reConditions.check(memberState.getGroup().equals(request.getGroup()), DLedgerResponseCode.UNKNOWN_GROUP, "%s != %s", request.getGroup(), memberState.getGroup());
3PreConditions.check(memberState.isLeader(), DLedgerResponseCode.NOT_LEADER);

Step1：首先验证请求的合理性：

如果请求的节点ID不是当前处理节点，则抛出异常。
如果请求的集群不是当前节点所在的集群，则抛出异常。
如果当前节点不是主节点，则抛出异常。

DLedgerServer#handleAppend

 1long currTerm = memberState.currTerm();
 2if (dLedgerEntryPusher.isPendingFull(currTerm)) {  // @1
 3    AppendEntryResponse appendEntryResponse = new AppendEntryResponse();
 4    appendEntryResponse.setGroup(memberState.getGroup());
 5    appendEntryResponse.setCode(DLedgerResponseCode.LEADER_PENDING_FULL.getCode());
 6    appendEntryResponse.setTerm(currTerm);
 7    appendEntryResponse.setLeaderId(memberState.getSelfId());
 8    return AppendFuture.newCompletedFuture(-1, appendEntryResponse);
 9} else {   // @2
10    DLedgerEntry dLedgerEntry = new DLedgerEntry();
11    dLedgerEntry.setBody(request.getBody());
12    DLedgerEntry resEntry = dLedgerStore.appendAsLeader(dLedgerEntry);
13    return dLedgerEntryPusher.waitAck(resEntry);
14}

Step2：如果预处理队列已经满了，则拒绝客户端请求，返回 LEADER_PENDING_FULL 错误码；如果未满，将请求封装成 DledgerEntry，则调用 dLedgerStore 方法追加日志，并且通过使用 dLedgerEntryPusher 的 waitAck 方法同步等待副本节点的复制响应，并最终将结果返回给调用方法。

代码@1：如果 dLedgerEntryPusher 的 push 队列已满，则返回追加一次，其错误码为 LEADER_PENDING_FULL。
代码@2：追加消息到 Leader 服务器，并向从节点广播，在指定时间内如果未收到从节点的确认，则认为追加失败。

接下来就按照上述三个要点进行展开：

判断 Push 队列是否已满
Leader 节点存储消息
主节点等待从节点复制 ACK

1.1 如何判断 Push 队列是否已满

DLedgerEntryPusher#isPendingFull

1public boolean isPendingFull(long currTerm) {
2    checkTermForPendingMap(currTerm, "isPendingFull");     // @1
3    return pendingAppendResponsesByTerm.get(currTerm).size() > dLedgerConfig.getMaxPendingRequestsNum(); // @2
4}

主要分两个步骤：
代码@1：检查当前投票轮次是否在 PendingMap 中，如果不在，则初始化，其结构为：Map< Long/ 投票轮次/, ConcurrentMap>>。

代码@2：检测当前等待从节点返回结果的个数是否超过其最大请求数量，可通过maxPendingRequests
Num 配置，该值默认为：10000。

上述逻辑比较简单，但疑问随着而来，ConcurrentMap> 中的数据是从何而来的呢？我们不妨接着往下看。

1.2 Leader 节点存储数据

Leader 节点的数据存储主要由 DLedgerStore 的 appendAsLeader 方法实现。DLedger 分别实现了基于内存、基于文件的存储实现，本文重点关注基于文件的存储实现，其实现类为：DLedgerMmapFileStore。

下面重点来分析一下数据存储流程，其入口为DLedgerMmapFileStore 的 appendAsLeader 方法。

DLedgerMmapFileStore#appendAsLeader

1PreConditions.check(memberState.isLeader(), DLedgerResponseCode.NOT_LEADER);
2PreConditions.check(!isDiskFull, DLedgerResponseCode.DISK_FULL);

Step1：首先判断是否可以追加数据，其判断依据主要是如下两点：

当前节点的状态是否是 Leader，如果不是，则抛出异常。
当前磁盘是否已满，其判断依据是 DLedger 的根目录或数据文件目录的使用率超过了允许使用的最大值，默认值为85%。

1ByteBuffer dataBuffer = localEntryBuffer.get();
2ByteBuffer indexBuffer = localIndexBuffer.get();

Step2：从本地线程变量获取一个数据与索引 buffer。其中用于存储数据的 ByteBuffer，其容量固定为 4M ，索引的 ByteBuffer 为两个索引条目的长度，固定为64个字节。

 1DLedgerEntryCoder.encode(entry, dataBuffer);
 2public static void encode(DLedgerEntry entry, ByteBuffer byteBuffer) {
 3    byteBuffer.clear();
 4    int size = entry.computSizeInBytes();
 5    //always put magic on the first position
 6    byteBuffer.putInt(entry.getMagic());
 7    byteBuffer.putInt(size);
 8    byteBuffer.putLong(entry.getIndex());
 9    byteBuffer.putLong(entry.getTerm());
10    byteBuffer.putLong(entry.getPos());
11    byteBuffer.putInt(entry.getChannel());
12    byteBuffer.putInt(entry.getChainCrc());
13    byteBuffer.putInt(entry.getBodyCrc());
14    byteBuffer.putInt(entry.getBody().length);
15    byteBuffer.put(entry.getBody());
16    byteBuffer.flip();
17}

Step3：将 DLedgerEntry，即将数据写入到 ByteBuffer中，从这里看出，每一次写入会调用 ByteBuffer 的 clear 方法，将数据清空，从这里可以看出，每一次数据追加，只能存储4M的数据。

DLedgerMmapFileStore#appendAsLeader

1synchronized (memberState) {
2    PreConditions.check(memberState.isLeader(), DLedgerResponseCode.NOT_LEADER, null);
3    // ... 省略代码
4}

Step4：锁定状态机，并再一次检测节点的状态是否是 Leader 节点。

DLedgerMmapFileStore#appendAsLeader

1long nextIndex = ledgerEndIndex + 1;
2entry.setIndex(nextIndex);
3entry.setTerm(memberState.currTerm());
4entry.setMagic(CURRENT_MAGIC);
5DLedgerEntryCoder.setIndexTerm(dataBuffer, nextIndex, memberState.currTerm(), CURRENT_MAGIC);

Step5：为当前日志条目设置序号，即 entryIndex 与 entryTerm (投票轮次)。并将魔数、entryIndex、entryTerm 等写入到 bytebuffer 中。

DLedgerMmapFileStore#appendAsLeader

1long prePos = dataFileList.preAppend(dataBuffer.remaining());
2entry.setPos(prePos);
3PreConditions.check(prePos != -1, DLedgerResponseCode.DISK_ERROR, null);
4DLedgerEntryCoder.setPos(dataBuffer, prePos);

Step6：计算新的消息的起始偏移量，关于 dataFileList 的 preAppend 后续详细介绍其实现，然后将该偏移量写入日志的 bytebuffer 中。

DLedgerMmapFileStore#appendAsLeader

1for (AppendHook writeHook : appendHooks) {
2    writeHook.doHook(entry, dataBuffer.slice(), DLedgerEntry.BODY_OFFSET);
3}

Step7：执行钩子函数。

DLedgerMmapFileStore#appendAsLeader

1long dataPos = dataFileList.append(dataBuffer.array(), 0, dataBuffer.remaining());
2PreConditions.check(dataPos != -1, DLedgerResponseCode.DISK_ERROR, null);
3PreConditions.check(dataPos == prePos, DLedgerResponseCode.DISK_ERROR, null);

Step8：将数据追加到 pagecache 中。该方法稍后详细介绍。

DLedgerMmapFileStore#appendAsLeader

1DLedgerEntryCoder.encodeIndex(dataPos, entrySize, CURRENT_MAGIC, nextIndex, memberState.currTerm(), indexBuffer);
2long indexPos = indexFileList.append(indexBuffer.array(), 0, indexBuffer.remaining(), false);
3PreConditions.check(indexPos == entry.getIndex() * INDEX_UNIT_SIZE, DLedgerResponseCode.DISK_ERROR, null);

Step9：构建条目索引并将索引数据追加到 pagecache。

DLedgerMmapFileStore#appendAsLeader

1ledgerEndIndex++;
2ledgerEndTerm = memberState.currTerm();
3if (ledgerBeginIndex == -1) {
4    ledgerBeginIndex = ledgerEndIndex;
5}
6updateLedgerEndIndexAndTerm();

Step10：ledgerEndeIndex 加一（下一个条目）的序号。并设置 leader 节点的状态机的 ledgerEndIndex 与 ledgerEndTerm。

Leader 节点数据追加就介绍到这里，稍后会重点介绍与存储相关方法的实现细节。

1.3 主节点等待从节点复制 ACK

其实现入口为 dLedgerEntryPusher 的 waitAck 方法。

DLedgerEntryPusher#waitAck

 1public CompletableFuture<AppendEntryResponse> waitAck(DLedgerEntry entry) {
 2    updatePeerWaterMark(entry.getTerm(), memberState.getSelfId(), entry.getIndex());    // @1
 3    if (memberState.getPeerMap().size() == 1) {                                                                  // @2
 4        AppendEntryResponse response = new AppendEntryResponse();
 5        response.setGroup(memberState.getGroup());
 6        response.setLeaderId(memberState.getSelfId());
 7        response.setIndex(entry.getIndex());
 8        response.setTerm(entry.getTerm());
 9        response.setPos(entry.getPos());
10        return AppendFuture.newCompletedFuture(entry.getPos(), response);
11    } else {
12        checkTermForPendingMap(entry.getTerm(), "waitAck");                                            
13        AppendFuture<AppendEntryResponse> future = new AppendFuture<>(dLedgerConfig.getMaxWaitAckTimeMs()); // @3
14        future.setPos(entry.getPos());
15        CompletableFuture<AppendEntryResponse> old = pendingAppendResponsesByTerm.get(entry.getTerm()).put(entry.getIndex(), future);     // @4
16        if (old != null) {
17            logger.warn("[MONITOR] get old wait at index={}", entry.getIndex());
18        }
19        wakeUpDispatchers();                                       // @5
20        return future;
21    }
22}

代码@1：更新当前节点的 push 水位线。

代码@2：如果集群的节点个数为1，无需转发，直接返回成功结果。

代码@3：构建 append 响应 Future 并设置超时时间，默认值为：2500 ms，可以通过 maxWaitAckTimeMs 配置改变其默认值。

代码@4：将构建的 Future 放入等待结果集合中。

代码@5：唤醒 Entry 转发线程，即将主节点中的数据 push 到各个从节点。

接下来分别对上述几个关键点进行解读。

1.3.1 updatePeerWaterMark 方法

DLedgerEntryPusher#updatePeerWaterMark

1private void updatePeerWaterMark(long term, String peerId, long index) {    // 代码@1
2    synchronized (peerWaterMarksByTerm) { 
3       checkTermForWaterMark(term, "updatePeerWaterMark");                     // 代码@2
4        if (peerWaterMarksByTerm.get(term).get(peerId) < index) {                   // 代码@3
5            peerWaterMarksByTerm.get(term).put(peerId, index);
6        }
7    }
8}

代码@1：先来简单介绍该方法的两个参数：

long term
当前的投票轮次。
String peerId
当前节点的ID。
long index
当前追加数据的序号。

代码@2：初始化 peerWaterMarksByTerm 数据结构，其结果为 < Long \/ term */, Map< String \/* peerId /, Long \/ entry index*/>。

代码@3：如果 peerWaterMarksByTerm 存储的 index 小于当前数据的 index，则更新。

1.3.2 wakeUpDispatchers 详解

DLedgerEntryPusher#updatePeerWaterMark

1public void wakeUpDispatchers() {
2    for (EntryDispatcher dispatcher : dispatcherMap.values()) {
3        dispatcher.wakeup();
4    }
5}

该方法主要就是遍历转发器并唤醒。本方法的核心关键就是 EntryDispatcher，在详细介绍它之前我们先来看一下该集合的初始化。

DLedgerEntryPusher 构造方法

1for (String peer : memberState.getPeerMap().keySet()) {
2    if (!peer.equals(memberState.getSelfId())) {
3        dispatcherMap.put(peer, new EntryDispatcher(peer, logger));
4    }
5}

原来在构建 DLedgerEntryPusher 时会为每一个从节点创建一个 EntryDispatcher 对象。

显然，日志的复制由 DLedgerEntryPusher 来实现。由于篇幅的原因，该部分内容将在下篇文章中继续。

上面在讲解 Leader 追加日志时并没有详细分析存储相关的实现，为了知识体系的完备，接下来我们来分析一下其核心实现。

2、日志存储实现详情

主要对 MmapFileList 的 preAppend 与 append 方法进行详细讲解。

存储部分的设计请查阅笔者的博客：源码分析RocketMQ DLedger 多副本存储实现，MmapFileList 对标 RocketMQ 的MappedFileQueue。

2.1 MmapFileList 的 preAppend 详解

该方法最终会调用两个参数的preAppend方法，故我们直接来看两个参数的 preAppend 方法。

MmapFileList#preAppend

 1public long preAppend(int len, boolean useBlank) {                // @1
 2    MmapFile mappedFile = getLastMappedFile();                   // @2 start
 3    if (null == mappedFile || mappedFile.isFull()) {
 4        mappedFile = getLastMappedFile(0);
 5    }
 6    if (null == mappedFile) {
 7        logger.error("Create mapped file for {}", storePath);
 8        return -1;
 9    }                                                                                            // @2 end
10    int blank = useBlank ? MIN_BLANK_LEN : 0;
11    if (len + blank > mappedFile.getFileSize() - mappedFile.getWrotePosition()) {   // @3
12        if (blank < MIN_BLANK_LEN) {
13            logger.error("Blank {} should ge {}", blank, MIN_BLANK_LEN);
14            return -1;
15        } else {
16            ByteBuffer byteBuffer = ByteBuffer.allocate(mappedFile.getFileSize() - mappedFile.getWrotePosition());     // @4
17            byteBuffer.putInt(BLANK_MAGIC_CODE);                                                                                                      // @5
18            byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition());                                               // @6
19            if (mappedFile.appendMessage(byteBuffer.array())) {                                                                                     // @7
20                //need to set the wrote position
21                mappedFile.setWrotePosition(mappedFile.getFileSize());
22            } else {
23                logger.error("Append blank error for {}", storePath);
24                return -1;
25            }
26            mappedFile = getLastMappedFile(0);
27            if (null == mappedFile) {
28                logger.error("Create mapped file for {}", storePath);
29                return -1;
30            }
31        }
32    }
33    return mappedFile.getFileFromOffset() + mappedFile.getWrotePosition();// @8
34}

代码@1：首先介绍其参数的含义：

int len 需要申请的长度。
boolean useBlank 是否需要填充，默认为true。
代码@2：获取最后一个文件，即获取当前正在写的文件。

代码@3：如果需要申请的资源超过了当前文件可写字节时，需要处理的逻辑。代码@4-@7都是其处理逻辑。

代码@4：申请一个当前文件剩余字节的大小的bytebuffer。

代码@5：先写入魔数。

代码@6：写入字节长度，等于当前文件剩余的总大小。

代码@7：写入空字节，代码@4-@7的用意就是写一条空Entry，填入魔数与 size，方便解析。

代码@8：如果当前文件足以容纳待写入的日志，则直接返回其物理偏移量。

经过上述代码解读，我们很容易得出该方法的作用，就是返回待写入日志的起始物理偏移量。

2.2 MmapFileList 的 append 详解

最终会调用4个参数的 append 方法，其代码如下：
MmapFileList#append

 1public long append(byte[] data, int pos, int len, boolean useBlank) {  // @1
 2    if (preAppend(len, useBlank) == -1) {
 3        return -1;
 4    }
 5    MmapFile mappedFile = getLastMappedFile();                               // @2
 6    long currPosition = mappedFile.getFileFromOffset() + mappedFile.getWrotePosition();   // @3
 7    if (!mappedFile.appendMessage(data, pos, len)) {            // @4
 8        logger.error("Append error for {}", storePath);
 9        return -1;
10    }
11    return currPosition;
12}

代码@1：首先介绍一下各个参数：

byte[] data
待写入的数据，即待追加的日志。
int pos
从 data 字节数组哪个位置开始读取。
int len
待写入的字节数量。
boolean useBlank
是否使用填充，默认为 true。

代码@2：获取最后一个文件，即当前可写的文件。

代码@3：获取当前写入指针。

代码@4：追加消息。

最后我们再来看一下 appendMessage，具体的消息追加实现逻辑。

DefaultMmapFile#appendMessage

 1public boolean appendMessage(final byte[] data, final int offset, final int length) {
 2    int currentPos = this.wrotePosition.get();
 3
 4    if ((currentPos + length) <= this.fileSize) {
 5        ByteBuffer byteBuffer = this.mappedByteBuffer.slice(); // @1
 6        byteBuffer.position(currentPos);
 7        byteBuffer.put(data, offset, length);
 8        this.wrotePosition.addAndGet(length);
 9        return true;
10    }
11    return false;
12}

该方法主要突出一下写入的方式是 mappedByteBuffer，是通过 FileChannel 的 map方法创建，即我们常说的 PageCache，即消息追加首先是写入到 pageCache 中。

本文详细介绍了 Leader 节点处理客户端消息追加请求的前面两个步骤，即判断 Push 队列是否已满与 Leader 节点存储消息。考虑到篇幅的问题，各个节点的数据同步将在下一篇文章中详细介绍。

在进入下一篇的文章学习之前，我们不妨思考一下如下问题：

如果主节点追加成功（写入到 PageCache)，但同步到从节点过程失败或此时主节点宕机，集群中的数据如何保证一致性？
亲爱的读者朋友们，都读到这里了，麻烦帮忙个点【在看】，谢谢。

推荐阅读源码分析 RocketMQ DLedger 多副本系列文章：

1、RocketMQ 多副本前置篇：初探raft协议

2、源码分析RocketMQ多副本之Leader选主

3、源码分析 RocketMQ DLedger 多副本存储实现

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

源码分析 RocketMQ DLedger(多副本) 之日志追加流程

标签：mappedFile,代码,byteBuffer,DLedger,源码,memberState,entry,节点,RocketMQ
来源： https://blog.51cto.com/15023237/2558978