基于Raft分布式Kv存储：doHeartBeat

doHeartBeat()并不只是发送“我还活着”的空心跳它实际上是 Leader 的一次完整日志同步调度给每个 Follower 构造AppendEntries。落后太多的节点改发快照。并发发送 RPC。根据回复更新nextIndex、matchIndex和commitIndex。遇到更高任期时退回 Follower。所以更准确的名字其实接近replicateLogsToFollowers()。调用链它有两个主要入口节点刚当选 Leader └── 立即启动 doHeartBeat() leaderHearBeatTicker() └── 每隔 HeartBeatTimeout 调用 doHeartBeat()当选后立即调用是为了尽快向其他节点宣布 Leader 身份之后由心跳定时器周期触发。doHeartBeat()本身不等待 RPC 返回而是为每个节点启动一个分离线程执行sendAppendEntries()。整体伪代码lock(m_mtx); if (当前仍是 Leader) { 本轮成功节点数 1; // Leader 自己 for (每个其他节点 i) { if (i 落后到 Leader 已经压缩掉的日志之前) { 异步发送快照(i); continue; } 根据 nextIndex[i] 计算 prevLogIndex 和 prevLogTerm; 构造 AppendEntries 参数; 把 nextIndex[i] 到 Leader 最后一条日志装入 entries; 异步执行 sendAppendEntries(i, args, reply, 成功节点数); } 重置 Leader 心跳计时器; }一、加锁并检查身份std::lock_guardstd::mutex g(m_mtx); if (m_status Leader) { ... }整个请求构造过程都持有m_mtx。这样一轮心跳中读取到的currentTerm、日志、快照位置和commitIndex是一致的。即使定时器判断时还是 Leader在真正进入函数前也可能因为收到更高任期消息变成 Follower所以拿锁之后必须再次检查m_status。二、appendNums为什么从 1 开始auto appendNums std::make_sharedint(1);它表示这一轮AppendEntries成功的节点数量。从1开始是因为 Leader 自己天然已经拥有自己的日志不需要向自己发送 RPC。使用shared_ptr是因为doHeartBeat()很快就会返回但多个异步线程仍要共同使用这个计数器。计数器虽然不是atomic但当前实现中的加法发生在持有m_mtx的情况下因此加法本身受到互斥保护三、遍历所有 Followerfor (int i 0; i m_peers.size(); i) { if (i m_me) { continue; } }Leader 为每个 Follower 独立维护nextIndex[i] 下一次准备发给节点 i 的日志下标 matchIndex[i] 已确认节点 i 拥有的最大日志下标不同 Follower 的进度可能完全不同所以每个节点收到的AppendEntries参数也不同。四、判断应该发快照还是日志if (m_nextIndex[i] m_lastSnapshotIncludeIndex) { std::thread t(Raft::leaderSendSnapShot, this, i); t.detach(); continue; }假设 Leader 已经把1100的日志压缩成快照而某个 Follower 的nextIndex[i] 80Leader 内存里已经没有80100的普通日志无法继续通过AppendEntries补齐。因此只能发送包含这些历史状态的快照。快照成功后m_matchIndex[server] snapshotIndex; m_nextIndex[server] snapshotIndex 1;下一轮心跳再从快照之后继续发送普通日志。五、 计算prevLogIndex和prevLogTermgetPrevLogInfo(i, preLogIndex, PrevLogTerm);正常情况下preLogIndex m_nextIndex[i] - 1; preLogTerm term(preLogIndex);例如Leader 日志1 2 3 4 5 6 7 8 Follower 的 nextIndex 6那么请求会携带prevLogIndex 5 prevLogTerm Leader 第 5 条日志的 term entries 日志 6、7、8Follower 必须先确认自己的第 5 条日志和 Leader 的第 5 条日志任期相同才能接受后面的日志。这就是 Raft 的日志连续性检查。六、 构造AppendEntriesArgsappendEntriesArgs-set_term(m_currentTerm); appendEntriesArgs-set_leaderid(m_me); appendEntriesArgs-set_prevlogindex(preLogIndex); appendEntriesArgs-set_prevlogterm(PrevLogTerm); appendEntriesArgs-set_leadercommit(m_commitIndex);各字段含义term Leader 当前任期 leaderId Leader 节点编号 prevLogIndex 新日志前一条日志的下标 prevLogTerm 前一条日志的任期 entries 要追加的日志可以为空 leaderCommit Leader 已提交到哪里leaderCommit很重要。即使这次没有新日志也能通知 Follower“之前发送给你的日志现在已经提交到哪个位置了”。七、装入待同步日志代码从preLogIndex 1开始一直把日志装到 Leader 的最后一条for (...) { auto* entry appendEntriesArgs-add_entries(); *entry m_logs[j]; }因此这个实现中的“心跳”有两种形式Follower 已追平 entries 为空是真正的空心跳 Follower 落后 entries 包含缺少的日志同时完成日志复制源码中的断言prevLogIndex entries_size lastLogIndex保证本次请求确实从prevLogIndex后面一直覆盖到 Leader 日志末尾。八、 异步发送 RPCstd::thread t( Raft::sendAppendEntries, this, i, appendEntriesArgs, appendEntriesReply, appendNums ); t.detach();这里没有在doHeartBeat()内直接执行 RPC因为网络调用可能很慢。执行效果是doHeartBeat ├── 线程1 → Follower 1 ├── 线程2 → Follower 2 ├── 线程3 → Follower 3 └── 立即返回不等待结果shared_ptr保证请求参数、回复对象和计数器在异步线程结束前不会被释放。九、sendAppendEntries()如何处理回复RPC 调用时不持有 Raft 主锁bool ok m_peers[server]-AppendEntries(args.get(), reply.get());网络调用完成后才重新拿锁处理状态。处理顺序大致是网络失败 └── 直接返回 回复中的 term 更大 └── 当前 Leader 退回 Follower 回复 term 更小 └── 这是过期回复忽略 当前节点已经不是 Leader └── 忽略回复 success false └── 根据 Follower 建议回退 nextIndex success true ├── 更新 matchIndex ├── 更新 nextIndex └── 尝试推进 commitIndex成功时m_matchIndex[server] std::max(m_matchIndex[server], args-prevlogindex() args-entries_size()); m_nextIndex[server] m_matchIndex[server] 1;使用max是为了避免较旧的成功回复把已经前进的matchIndex再改小。十、Follower 收到心跳后发生什么Follower 的AppendEntries1()会拒绝任期过小的旧 Leader并且不重置选举计时器。如果请求任期更大更新任期并转为 Follower。同任期 Candidate 收到合法 Leader 的请求也转为 Follower。重置选举超时计时器。检查prevLogIndex和prevLogTerm。匹配成功后追加或修正日志。按min(leaderCommit, lastLogIndex)更新自己的提交位置。因此即使entries为空这个 RPC 仍然完成了维持 Leader 权威和传播提交位置的工作