源码分析 RocketMQ DLedger(多副本) 之日志复制-上篇
本文紧接着 源码分析 RocketMQ DLedger(多副本) 之日志追加流程 ，继续 Leader 处理客户端 append 的请求流程中最至关重要的一环：日志复制。
温馨提示：由于微信单篇文章的字数限制，RocketMQ DLedger 日志复制分为两篇文章介绍。
DLedger 多副本的日志转发由 DLedgerEntryPusher 实现，接下来将对其进行详细介绍。
温馨提示：由于本篇幅较长，为了更好的理解其实现，大家可以带着如下疑问来通读本篇文章：
1、raft 协议中有一个非常重要的概念：已提交日志序号，该如何实现。
2、客户端向 DLedger 集群发送一条日志，必须得到集群中大多数节点的认可才能被认为写入成功。
3、raft 协议中追加、提交两个动作如何实现。
日志复制(日志转发)由 DLedgerEntryPusher 实现，具体类图如下：
DLedgerEntryPusher
DLedger 日志转发与处理核心类，该内会启动如下3个对象，其分别对应一个线程。EntryHandler
日志接收处理线程，当节点为从节点时激活。QuorumAckChecker
日志追加ACK投票处理线程，当前节点为主节点时激活。EntryDispatcher
日志转发线程，当前节点为主节点时追加。
接下来我们将详细介绍上述4个类，从而揭晓日志复制的核心实现原理。
1、DLedgerEntryPusher
1.1 核心类图
DLedgerConfig dLedgerConfig
多副本相关配置。DLedgerStore dLedgerStore
存储实现类。MemberState memberState
节点状态机。DLedgerRpcService dLedgerRpcService
RPC 服务实现类，用于集群内的其他节点进行网络通讯。Map> peerWaterMarksByTerm
每个节点基于投票轮次的当前水位线标记。键值为投票轮次，值为 ConcurrentMap/, Long/* 节点对应的日志序号*/>。Map>> pendingAppendResponsesByTerm
用于存放追加请求的响应结果(Future模式)。EntryHandler entryHandler
从节点上开启的线程，用于接收主节点的 push 请求（append、commit、append）。QuorumAckChecker quorumAckChecker
主节点上的追加请求投票器。Map dispatcherMap
主节点日志请求转发器，向从节点复制消息等。
接下来介绍一下其核心方法的实现。
1.2 构造方法
1public DLedgerEntryPusher(DLedgerConfig dLedgerConfig, MemberState memberState, DLedgerStore dLedgerStore,
2 DLedgerRpcService dLedgerRpcService) {
3 this.dLedgerConfig = dLedgerConfig;
4 this.memberState = memberState;
5 this.dLedgerStore = dLedgerStore;
6 this.dLedgerRpcService = dLedgerRpcService;
7 for (String peer : memberState.getPeerMap().keySet()) {
8 if (!peer.equals(memberState.getSelfId())) {
9 dispatcherMap.put(peer, new EntryDispatcher(peer, logger));
10 }
11 }
12}
构造方法的重点是会根据集群内的节点，依次构建对应的 EntryDispatcher 对象。
1.3 startup
DLedgerEntryPusher#startup
1public void startup() {
2 entryHandler.start();
3 quorumAckChecker.start();
4 for (EntryDispatcher dispatcher : dispatcherMap.values()) {
5 dispatcher.start();
6 }
7}
依次启动 EntryHandler、QuorumAckChecker 与 EntryDispatcher 线程。
备注：DLedgerEntryPusher 的其他核心方法在详细分析其日志复制原理的过程中会一一介绍。
接下来将从 EntryDispatcher、QuorumAckChecker、EntryHandler 来阐述 RocketMQ DLedger(多副本)的实现原理。
2、EntryDispatcher 详解
2.1 核心类图
其核心属性如下。
AtomicReference type
向从节点发送命令的类型，可选值：PushEntryRequest.Type.COMPARE、TRUNCATE、APPEND、COMMIT，下面详细说明。long lastPushCommitTimeMs = -1
上一次发送提交类型的时间戳。String peerId
目标节点ID。long compareIndex = -1
已完成比较的日志序号。long writeIndex = -1
已写入的日志序号。int maxPendingSize = 1000
允许的最大挂起日志数量。long term = -1
Leader 节点当前的投票轮次。String leaderId = null
Leader 节点ID。long lastCheckLeakTimeMs
上次检测泄漏的时间，所谓的泄漏，就是看挂起的日志请求数量是否查过了 maxPendingSize 。ConcurrentMap pendingMap
记录日志的挂起时间，key：日志的序列(entryIndex)，value：挂起时间戳。Quota quota = new Quota(dLedgerConfig.getPeerPushQuota())
配额。
2.2 Push 请求类型
DLedger 主节点向从从节点复制日志总共定义了4类请求类型，其枚举类型为 PushEntryRequest.Type，其值分别为 COMPARE、TRUNCATE、APPEND、COMMIT。
COMPARE
如果 Leader 发生变化，新的 Leader 需要与他的从节点的日志条目进行比较，以便截断从节点多余的数据。TRUNCATE
如果 Leader 通过索引完成日志对比，则 Leader 将发送 TRUNCATE 给它的从节点。APPEND
将日志条目追加到从节点。COMMIT
通常，leader 会将提交的索引附加到 append 请求，但是如果 append 请求很少且分散，leader 将发送一个单独的请求来通知从节点提交的索引。
对主从节点的请求类型有了一个初步的认识后，我们将从 EntryDispatcher 的业务处理入口 doWork 方法开始讲解。
2.3 doWork 方法详解
1public void doWork() {
2 try {
3 if (!checkAndFreshState()) { // @1
4 waitForRunning(1);
5 return;
6 }
7
8 if (type.get() == PushEntryRequest.Type.APPEND) { // @2
9 doAppend();
10 } else {
11 doCompare(); // @3
12 }
13 waitForRunning(1);
14 } catch (Throwable t) {
15 DLedgerEntryPusher.logger.error("[Push-{}]Error in {} writeIndex={} compareIndex={}", peerId, getName(), writeIndex, compareIndex, t);
16 DLedgerUtils.sleep(500);
17 }
18}
代码@1：检查状态，是否可以继续发送 append 或 compare。
代码@2：如果推送类型为APPEND，主节点向从节点传播消息请求。
代码@3：主节点向从节点发送对比数据差异请求（当一个新节点被选举成为主节点时，往往这是第一步）。
2.3.1 checkAndFreshState 详解
EntryDispatcher#checkAndFreshState
1private boolean checkAndFreshState() {
2 if (!memberState.isLeader()) { // @1
3 return false;
4 }
5 if (term != memberState.currTerm() || leaderId == null || !leaderId.equals(memberState.getLeaderId())) { // @2
6 synchronized (memberState) {
7 if (!memberState.isLeader()) {
8 return false;
9 }
10 PreConditions.check(memberState.getSelfId().equals(memberState.getLeaderId()), DLedgerResponseCode.UNKNOWN);
11 term = memberState.currTerm();
12 leaderId = memberState.getSelfId();
13 changeState(-1, PushEntryRequest.Type.COMPARE);
14 }
15 }
16 return true;
17}
代码@1：如果节点的状态不是主节点，则直接返回 false。则结束 本次 doWork 方法。因为只有主节点才需要向从节点转发日志。
代码@2：如果当前节点状态是主节点，但当前的投票轮次与状态机轮次或 leaderId 还未设置，或 leaderId 与状态机的 leaderId 不相等，这种情况通常是集群触发了重新选举，设置其term、leaderId与状态机同步，即将发送COMPARE 请求。
接下来看一下 changeState (改变状态)。
1private synchronized void changeState(long index, PushEntryRequest.Type target) {
2 logger.info("[Push-{}]Change state from {} to {} at {}", peerId, type.get(), target, index);
3 switch (target) {
4 case APPEND: // @1
5 compareIndex = -1;
6 updatePeerWaterMark(term, peerId, index);
7 quorumAckChecker.wakeup();
8 writeIndex = index + 1;
9 break;
10 case COMPARE: // @2
11 if (this.type.compareAndSet(PushEntryRequest.Type.APPEND, PushEntryRequest.Type.COMPARE)) {
12 compareIndex = -1;
13 pendingMap.clear();
14 }
15 break;
16 case TRUNCATE: // @3
17 compareIndex = -1;
18 break;
19 default:
20 break;
21 }
22 type.set(target);
23}
代码@1：如果将目标类型设置为 append，则重置 compareIndex ，并设置 writeIndex 为当前 index 加1。
代码@2：如果将目标类型设置为 COMPARE，则重置 compareIndex 为负一，接下将向各个从节点发送 COMPARE 请求类似，并清除已挂起的请求。
代码@3：如果将目标类型设置为 TRUNCATE，则重置 compareIndex 为负一。
接下来具体来看一下 APPEND、COMPARE、TRUNCATE 等请求。
2.3.2 append 请求详解
EntryDispatcher#doAppend
1private void doAppend() throws Exception {
2 while (true) {
3 if (!checkAndFreshState()) { // @1
4 break;
5 }
6 if (type.get() != PushEntryRequest.Type.APPEND) { // @2
7 break;
8 }
9 if (writeIndex > dLedgerStore.getLedgerEndIndex()) { // @3
10 doCommit();
11 doCheckAppendResponse();
12 break;
13 }
14 if (pendingMap.size() >= maxPendingSize || (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000)) { // @4
15 long peerWaterMark = getPeerWaterMark(term, peerId);
16 for (Long index : pendingMap.keySet()) {
17 if (index < peerWaterMark) {
18 pendingMap.remove(index);
19 }
20 }
21 lastCheckLeakTimeMs = System.currentTimeMillis();
22 }
23 if (pendingMap.size() >= maxPendingSize) { // @5
24 doCheckAppendResponse();
25 break;
26 }
27 doAppendInner(writeIndex); // @6
28 writeIndex++;
29 }
30}
代码@1：检查状态，已经在上面详细介绍。
代码@2：如果请求类型不为 APPEND，则退出，结束本轮 doWork 方法执行。
代码@3：writeIndex 表示当前追加到从该节点的序号，通常情况下主节点向从节点发送 append 请求时，会附带主节点的已提交指针，但如何 append 请求发不那么频繁，writeIndex 大于 leaderEndIndex 时（由于pending请求超过其 pending 请求的队列长度（默认为1w)，时，会阻止数据的追加，此时有可能出现 writeIndex 大于 leaderEndIndex 的情况，此时单独发送 COMMIT 请求。
代码@4：检测 pendingMap(挂起的请求数量)是否发送泄漏，即挂起队列中容量是否超过允许的最大挂起阀值。获取当前节点关于本轮次的当前水位线(已成功 append 请求的日志序号)，如果发现正在挂起请求的日志序号小于水位线，则丢弃。
代码@5：如果挂起的请求（等待从节点追加结果）大于 maxPendingSize 时，检查并追加一次 append 请求。
代码@6：具体的追加请求。
2.3.2.1 doCommit 发送提交请求
EntryDispatcher#doCommit
1private void doCommit() throws Exception {
2 if (DLedgerUtils.elapsed(lastPushCommitTimeMs) > 1000) { // @1
3 PushEntryRequest request = buildPushRequest(null, PushEntryRequest.Type.COMMIT); // @2
4 //Ignore the results
5 dLedgerRpcService.push(request); // @3
6 lastPushCommitTimeMs = System.currentTimeMillis();
7 }
8}
代码@1：如果上一次单独发送 commit 的请求时间与当前时间相隔低于 1s，放弃本次提交请求。
代码@2：构建提交请求。
代码@3：通过网络向从节点发送 commit 请求。
接下来先了解一下如何构建 commit 请求包。
EntryDispatcher#buildPushRequest
1private PushEntryRequest buildPushRequest(DLedgerEntry entry, PushEntryRequest.Type target) {
2 PushEntryRequest request = new PushEntryRequest();
3 request.setGroup(memberState.getGroup());
4 request.setRemoteId(peerId);
5 request.setLeaderId(leaderId);
6 request.setTerm(term);
7 request.setEntry(entry);
8 request.setType(target);
9 request.setCommitIndex(dLedgerStore.getCommittedIndex());
10 return request;
11}
提交包请求字段主要包含如下字段：DLedger 节点所属组、从节点 id、主节点 id，当前投票轮次、日志内容、请求类型与 committedIndex(主节点已提交日志序号)。
2.3.2.2 doCheckAppendResponse 检查并追加请求
EntryDispatcher#doCheckAppendResponse
1private void doCheckAppendResponse() throws Exception {
2 long peerWaterMark = getPeerWaterMark(term, peerId); // @1
3 Long sendTimeMs = pendingMap.get(peerWaterMark + 1);
4 if (sendTimeMs != null && System.currentTimeMillis() - sendTimeMs > dLedgerConfig.getMaxPushTimeOutMs()) { // @2
5 logger.warn("[Push-{}]Retry to push entry at {}", peerId, peerWaterMark + 1);
6 doAppendInner(peerWaterMark + 1);
7 }
8}
该方法的作用是检查 append 请求是否超时，其关键实现如下：
获取已成功 append 的序号。
从挂起的请求队列中获取下一条的发送时间，如果不为空并去超过了 append 的超时时间，则再重新发送 append 请求，最大超时时间默认为 1s，可以通过 maxPushTimeOutMs 来改变默认值。
2.3.2.3 doAppendInner 追加请求
向从节点发送 append 请求。
EntryDispatcher#doAppendInner
1private void doAppendInner(long index) throws Exception {
2 DLedgerEntry entry = dLedgerStore.get(index); // @1
3 PreConditions.check(entry != null, DLedgerResponseCode.UNKNOWN, "writeIndex=%d", index);
4 checkQuotaAndWait(entry); // @2
5 PushEntryRequest request = buildPushRequest(entry, PushEntryRequest.Type.APPEND); // @3
6 CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(request); // @4
7 pendingMap.put(index, System.currentTimeMillis()); // @5
8 responseFuture.whenComplete((x, ex) -> {
9 try {
10 PreConditions.check(ex == null, DLedgerResponseCode.UNKNOWN);
11 DLedgerResponseCode responseCode = DLedgerResponseCode.valueOf(x.getCode());
12 switch (responseCode) {
13 case SUCCESS: // @6
14 pendingMap.remove(x.getIndex());
15 updatePeerWaterMark(x.getTerm(), peerId, x.getIndex());
16 quorumAckChecker.wakeup();
17 break;
18 case INCONSISTENT_STATE: // @7
19 logger.info("[Push-{}]Get INCONSISTENT_STATE when push index={} term={}", peerId, x.getIndex(), x.getTerm());
20 changeState(-1, PushEntryRequest.Type.COMPARE);
21 break;
22 default:
23 logger.warn("[Push-{}]Get error response code {} {}", peerId, responseCode, x.baseInfo());
24 break;
25 }
26 } catch (Throwable t) {
27 logger.error("", t);
28 }
29 });
30 lastPushCommitTimeMs = System.currentTimeMillis();
31}
代码@1：首先根据序号查询出日志。
代码@2：检测配额，如果超过配额，会进行一定的限流，其关键实现点：
首先触发条件：append 挂起请求数已超过最大允许挂起数；基于文件存储并主从差异超过300m，可通过 peerPushThrottlePoint 配置。
每秒追加的日志超过 20m(可通过 peerPushQuota 配置)，则会 sleep 1s中后再追加。
代码@3：构建 PUSH 请求日志。
代码@4：通过 Netty 发送网络请求到从节点，从节点收到请求会进行处理(本文并不会探讨与网络相关的实现细节)。
代码@5：用 pendingMap 记录待追加的日志的发送时间，用于发送端判断是否超时的一个依据。
代码@6：请求成功的处理逻辑，其关键实现点如下：
移除 pendingMap 中的关于该日志的发送超时时间。
更新已成功追加的日志序号(按投票轮次组织，并且每个从服务器一个键值对)。
唤醒 quorumAckChecker 线程(主要用于仲裁 append 结果)，后续会详细介绍。
代码@7：Push 请求出现状态不一致情况，将发送 COMPARE 请求，来对比主从节点的数据是否一致。
日志转发 append 追加请求类型就介绍到这里了，接下来我们继续探讨另一个请求类型 compare。
2.3.3 compare 请求详解
COMPARE 类型的请求有 doCompare 方法发送，首先该方法运行在 while (true) 中，故在查阅下面代码时，要注意其退出循环的条件。
EntryDispatcher#doCompare
1if (!checkAndFreshState()) {
2 break;
3}
4if (type.get() != PushEntryRequest.Type.COMPARE
5 && type.get() != PushEntryRequest.Type.TRUNCATE) {
6 break;
7}
8if (compareIndex == -1 && dLedgerStore.getLedgerEndIndex() == -1) {
9 break;
10}
Step1：验证是否执行，有几个关键点如下：
判断是否是主节点，如果不是主节点，则直接跳出。
如果是请求类型不是 COMPARE 或 TRUNCATE 请求，则直接跳出。
如果已比较索引 和 ledgerEndIndex 都为 -1 ，表示一个新的 DLedger 集群，则直接跳出。
EntryDispatcher#doCompare
1if (compareIndex == -1) {
2 compareIndex = dLedgerStore.getLedgerEndIndex();
3 logger.info("[Push-{}][DoCompare] compareIndex=-1 means start to compare", peerId);
4} else if (compareIndex > dLedgerStore.getLedgerEndIndex() || compareIndex < dLedgerStore.getLedgerBeginIndex()) {
5 logger.info("[Push-{}][DoCompare] compareIndex={} out of range {}-{}", peerId, compareIndex, dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex());
6 compareIndex = dLedgerStore.getLedgerEndIndex();
7}
Step2：如果 compareIndex 为 -1 或compareIndex 不在有效范围内，则重置待比较序列号为当前已已存储的最大日志序号：ledgerEndIndex。
1DLedgerEntry entry = dLedgerStore.get(compareIndex);
2PreConditions.check(entry != null, DLedgerResponseCode.INTERNAL_ERROR, "compareIndex=%d", compareIndex);
3PushEntryRequest request = buildPushRequest(entry, PushEntryRequest.Type.COMPARE);
4CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(request);
5PushEntryResponse response = responseFuture.get(3, TimeUnit.SECONDS);
Step3：根据序号查询到日志，并向从节点发起 COMPARE 请求，其超时时间为 3s。
EntryDispatcher#doCompare
1long truncateIndex = -1;
2if (response.getCode() == DLedgerResponseCode.SUCCESS.getCode()) { // @1
3 if (compareIndex == response.getEndIndex()) {
4 changeState(compareIndex, PushEntryRequest.Type.APPEND);
5 break;
6 } else {
7 truncateIndex = compareIndex;
8 }
9
10} else if (response.getEndIndex() < dLedgerStore.getLedgerBeginIndex()
11 || response.getBeginIndex() > dLedgerStore.getLedgerEndIndex()) { // @2
12 truncateIndex = dLedgerStore.getLedgerBeginIndex();
13} else if (compareIndex < response.getBeginIndex()) { // @3
14 truncateIndex = dLedgerStore.getLedgerBeginIndex();
15} else if (compareIndex > response.getEndIndex()) { // @4
16 compareIndex = response.getEndIndex();
17} else { // @5
18 compareIndex--;
19}
20
21if (compareIndex < dLedgerStore.getLedgerBeginIndex()) { // @6
22 truncateIndex = dLedgerStore.getLedgerBeginIndex();
23}
Step4：根据响应结果计算需要截断的日志序号，其主要实现关键点如下：
代码@1：如果两者的日志序号相同，则无需截断，下次将直接先从节点发送 append 请求；否则将 truncateIndex 设置为响应结果中的 endIndex。
代码@2：如果从节点存储的最大日志序号小于主节点的最小序号，或者从节点的最小日志序号大于主节点的最大日志序号，即两者不相交，这通常发生在从节点崩溃很长一段时间，而主节点删除了过期的条目时。truncateIndex 设置为主节点的 ledgerBeginIndex，即主节点目前最小的偏移量。
代码@3：如果已比较的日志序号小于从节点的开始日志序号，很可能是从节点磁盘发送损耗，从主节点最小日志序号开始同步。
代码@4：如果已比较的日志序号大于从节点的最大日志序号，则已比较索引设置为从节点最大的日志序号，触发数据的继续同步。
代码@5：如果已比较的日志序号大于从节点的开始日志序号，但小于从节点的最大日志序号，则待比较索引减一。
代码@6：如果比较出来的日志序号小于主节点的最小日志需要，则设置为主节点的最小序号。
1if (truncateIndex != -1) {
2 changeState(truncateIndex, PushEntryRequest.Type.TRUNCATE);
3 doTruncate(truncateIndex);
4 break;
5}
Step5：如果比较出来的日志序号不等于 -1 ，则向从节点发送 TRUNCATE 请求。
2.3.3.1 doTruncate 详解
1private void doTruncate(long truncateIndex) throws Exception {
2 PreConditions.check(type.get() == PushEntryRequest.Type.TRUNCATE, DLedgerResponseCode.UNKNOWN);
3 DLedgerEntry truncateEntry = dLedgerStore.get(truncateIndex);
4 PreConditions.check(truncateEntry != null, DLedgerResponseCode.UNKNOWN);
5 logger.info("[Push-{}]Will push data to truncate truncateIndex={} pos={}", peerId, truncateIndex, truncateEntry.getPos());
6 PushEntryRequest truncateRequest = buildPushRequest(truncateEntry, PushEntryRequest.Type.TRUNCATE);
7 PushEntryResponse truncateResponse = dLedgerRpcService.push(truncateRequest).get(3, TimeUnit.SECONDS);
8 PreConditions.check(truncateResponse != null, DLedgerResponseCode.UNKNOWN, "truncateIndex=%d", truncateIndex);
9 PreConditions.check(truncateResponse.getCode() == DLedgerResponseCode.SUCCESS.getCode(), DLedgerResponseCode.valueOf(truncateResponse.getCode()), "truncateIndex=%d", truncateIndex);
10 lastPushCommitTimeMs = System.currentTimeMillis();
11 changeState(truncateIndex, PushEntryRequest.Type.APPEND);
12}
该方法主要就是构建 truncate 请求到从节点。
关于服务端的消息复制转发就介绍到这里了，主节点负责向从服务器PUSH请求，从节点自然而然的要处理这些请求，接下来我们就按照主节点发送的请求，来具体分析一下从节点是如何响应的。
由于微信单篇文章字数的限制，从服务器接收到主节点的 PUSH 请求后如何处理、以及主服务根据所有从服务器的响应后进行仲裁(需要集群内半数以上节点追加成功后才认为是有效数据)等实现细节，则在下一篇文章中给出。
笔者推荐阅读 RocketMQ DLedger 多副本系列文章：
3、源码分析 RocketMQ DLedger 多副本存储实现
更多文章请关注中间件兴趣圈公众号：