sendAppendEntries()可以理解為針對某一個 Follower執(zhí)行一次AppendEntries RPC然后把回復(fù)合并回 Leader 的 Raft 狀態(tài)。它不負(fù)責(zé)構(gòu)造請求。請求由doHeartBeat()提前構(gòu)造它只負(fù)責(zé)發(fā)送 RPC ↓ 等待結(jié)果 ↓ 檢查任期和 Leader 身份 ↓ 失敗回退 nextIndex 成功推進 matchIndex、nextIndex ↓ 嘗試推進 commitIndex源碼實現(xiàn)可見于該項目的raft.cpp。函數(shù)參數(shù)函數(shù)大致接收四個參數(shù)void Raft::sendAppendEntries( int server, std::shared_ptrAppendEntriesArgs args, std::shared_ptrAppendEntriesReply reply, std::shared_ptrint appendNums);含義分別是server 目標(biāo) Follower 的編號。 args doHeartBeat() 構(gòu)造好的 AppendEntries 請求。 包含 term、prevLogIndex、prevLogTerm、entries、leaderCommit。 reply 保存 Follower 返回的響應(yīng)。 appendNums 本輪心跳中共享的“成功節(jié)點數(shù)量”。 doHeartBeat() 創(chuàng)建它時通常初始化為 1代表 Leader 自己。使用shared_ptr是因為函數(shù)運行在detach()出去的線程中。doHeartBeat()返回后請求、回復(fù)和計數(shù)器仍必須存活。一、發(fā)送 RPC核心調(diào)用類似bool ok m_peers[server]-AppendEntries( args.get(), reply.get() );這里值得注意的是執(zhí)行網(wǎng)絡(luò) RPC 時沒有持有m_mtx。這是正確的鎖邊界設(shè)計。網(wǎng)絡(luò)請求可能超時或阻塞如果在 RPC 期間一直持有 Raft 主鎖當(dāng)前節(jié)點將無法及時處理其他節(jié)點發(fā)來的 RPC接收更高任期處理客戶端請求更新選舉和心跳狀態(tài)。因此它采用無鎖執(zhí)行網(wǎng)絡(luò)調(diào)用 ↓ RPC 返回 ↓ 加鎖處理回復(fù)但這也意味著 RPC 飛行期間當(dāng)前節(jié)點的任期和身份可能發(fā)生變化所以后面必須重新校驗。二、網(wǎng)絡(luò)失敗直接返回if (!ok) { return; }ok false通常表示連接失敗、超時或者底層 RPC 沒有成功完成。這種情況下函數(shù)不會修改 nextIndex 修改 matchIndex 修改 currentTerm 增加 appendNums它也不會立即重試。后續(xù)的周期性doHeartBeat()會再次嘗試。這是合理的因為網(wǎng)絡(luò)失敗不代表日志不匹配不能因為一次超時就隨意回退nextIndex。三、重新獲得 Raft 鎖std::lock_guardstd::mutex lock(m_mtx);從這里開始回復(fù)處理期間的共享狀態(tài)修改都受同一把鎖保護包括m_currentTerm m_status m_votedFor m_nextIndex m_matchIndex m_commitIndex appendNums因此appendNums雖然只是普通int而不是原子變量但它的讀寫發(fā)生在m_mtx內(nèi)當(dāng)前實現(xiàn)下不會因為多個回復(fù)線程同時執(zhí)行而產(chǎn)生直接的數(shù)據(jù)競爭。四、處理更大的任期if (reply-term() m_currentTerm) { m_status Follower; m_currentTerm reply-term(); m_votedFor -1; return; }這是 Raft 非常重要的一條規(guī)則任何節(jié)點只要發(fā)現(xiàn)其他節(jié)點的任期比自己大就必須更新任期并退回 Follower。例如當(dāng)前節(jié)點認(rèn)為 自己是 term6 的 Leader Follower 回復(fù) term7這說明 term 6 已經(jīng)過期集群至少已經(jīng)進入 term 7。當(dāng)前節(jié)點不能再繼續(xù)發(fā)送 term 6 的日志必須立即退位。這里同時清空m_votedFor -1;表示新任期中還沒有投票。不過這份實現(xiàn)此處分支沒有明顯調(diào)用persist()。由于currentTerm和votedFor屬于 Raft 的持久化狀態(tài)更嚴(yán)格的實現(xiàn)應(yīng)該在釋放鎖或返回之前將它們持久化否則節(jié)點崩潰重啟后可能恢復(fù)出舊任期。五、丟棄較小任期的回復(fù)else if (reply-term() m_currentTerm) { return; }例如發(fā)送請求時 currentTerm 6 等待 RPC 期間 當(dāng)前節(jié)點進入 term 7并且重新成為 Leader 舊請求返回 reply.term 6這個回復(fù)屬于過去的任期不能再修改當(dāng)前狀態(tài)所以直接丟棄。隨后通常還有斷言assert(reply-term() m_currentTerm);經(jīng)過前面兩個分支后繼續(xù)執(zhí)行的回復(fù)原則上必須與當(dāng)前任期一致。六、再次確認(rèn)自己仍然是 Leaderif (m_status ! Leader) { return; }即使回復(fù)的任期等于m_currentTerm也不能說明當(dāng)前節(jié)點仍然是 Leader。RPC 飛行期間可能發(fā)生Leader 發(fā)送 AppendEntries ↓ 收到合法的同任期 Leader 消息或狀態(tài)發(fā)生變化 ↓ 當(dāng)前節(jié)點轉(zhuǎn)為 Follower ↓ 舊 AppendEntries 回復(fù)返回此時不能再更新 Leader 專屬的nextIndex[] matchIndex[] commitIndex所以需要獨立檢查m_status。七、處理日志匹配失敗if (!reply-success()) { if (reply-updatenextindex() ! -100) { m_nextIndex[server] reply-updatenextindex(); } }success false一般說明Follower 不存在 prevLogIndex或者Follower 在 prevLogIndex 位置的 term 和 Leader 給出的 prevLogTerm 不同F(xiàn)ollower 會通過updateNextIndex告訴 Leader下次應(yīng)該從哪里嘗試。例如Leader nextIndex[F] 8 本次發(fā)送 prevLogIndex 7 Follower 實際只有日志 14Follower 可以回復(fù)success false updateNextIndex 5Leader于是執(zhí)行m_nextIndex[F] 5;下一輪請求變成prevLogIndex 4 entries [5, 6, 7, ...]這就是 Raft 的日志回退過程。-100是這份代碼使用的特殊哨兵值表示 Follower 沒有提供可用的新下標(biāo)。工程上更清晰的方式是使用 Protobuf 的字段存在性或明確的狀態(tài)枚舉而不是魔法數(shù)字。八、成功時更新復(fù)制進度成功分支首先增加本輪成功數(shù)*appendNums *appendNums 1;然后計算這次請求能夠確認(rèn)的最大日志下標(biāo)int replicatedIndex args-prevlogindex() args-entries_size();例如prevLogIndex 5 entries [6, 7, 8] entries_size 3那么replicatedIndex 5 3 8意味著 Follower 已經(jīng)確認(rèn)擁有截至日志 8 的完整前綴。為什么更新matchIndex要用max代碼類似m_matchIndex[server] std::max( m_matchIndex[server], args-prevlogindex() args-entries_size() );原因是網(wǎng)絡(luò)回復(fù)可能亂序。假設(shè)同時存在兩個請求請求 A確認(rèn)到日志 8 請求 B確認(rèn)到日志 12如果 B 先回來matchIndex 12隨后舊請求 A 才回來。如果直接賦值就會錯誤地變成matchIndex 8使用max可以保證matchIndex 只能前進不能后退這是成功回復(fù)處理里做得比較穩(wěn)妥的地方。更新nextIndexm_nextIndex[server] m_matchIndex[server] 1;兩個字段的關(guān)系是matchIndex[i] 已確認(rèn) Follower i 擁有的最后日志下標(biāo) nextIndex[i] 下一次應(yīng)從哪個下標(biāo)繼續(xù)發(fā)送如果已經(jīng)確認(rèn) Follower 擁有到日志 8matchIndex 8 nextIndex 9下一次心跳如果 Leader 沒有新日志就會發(fā)送prevLogIndex 8 entries 空這就是純心跳。九、嘗試推進commitIndex當(dāng)前實現(xiàn)使用if (*appendNums 1 m_peers.size() / 2) { *appendNums 0; if (args-entries_size() 0) { m_commitIndex std::max(m_commitIndex, m_matchIndex[server]); } }假設(shè)有 5 個節(jié)點多數(shù)派數(shù)量 1 5 / 2 3appendNums初始是 1因為 Leader 自己已經(jīng)有日志。收到兩個 Follower 的成功回復(fù)后appendNums 3于是代碼認(rèn)為獲得多數(shù)派可以推進提交位置。設(shè)置成0是為了避免本輪后續(xù)回復(fù)再次觸發(fā)提交邏輯。這里存在一個重要正確性問題appendNums只統(tǒng)計“RPC 成功了幾個”但不同 Follower 成功確認(rèn)的日志位置可能不同。例如 5 節(jié)點集群Leader擁有日志到 10 Follower A成功確認(rèn)到 5 Follower B成功確認(rèn)到 10成功數(shù)量是Leader A B 3已經(jīng)過半如果 B 的回復(fù)正好讓appendNums達到 3當(dāng)前代碼可能執(zhí)行commitIndex 10但日志 10 實際只有Leader Follower B只有兩個節(jié)點并沒有過半。Follower A 只擁有到日志 5。正確算法應(yīng)該針對每個候選下標(biāo)N統(tǒng)計有多少節(jié)點滿足 matchIndex[i] N只有滿足多數(shù)節(jié)點的 matchIndex N 并且 log[N].term currentTerm才能把commitIndex推進到N。這也是 Raft 論文描述的 Leader 提交規(guī)則。(usenix.org)該項目其實已經(jīng)存在類似的leaderUpdateCommitIndex()回復(fù)成功后調(diào)用它會比appendNums更符合 Raft 語義。