第 58 章 / 共 58 章 卷十 代码讲解

Raft 共识模拟

分布式系统中，多个节点如何就"下一个操作是什么"达成一致？这就是共识算法要解决的问题。Raft（2014 年 Diego Ongaro 提出）是目前最易理解的共识算法——etcd、Consul、TiKV 都基于它。本章用 xlang 模拟一个 5 节点 Raft 集群：Leader 选举、日志复制、提交应用——全书的收官之作。

本章导读

Raft 将共识问题分解为三个子问题：① Leader 选举——集群中选出一个领导者；② 日志复制——Leader 将操作日志同步给所有节点；③ 安全性——保证所有节点最终执行相同的操作序列。本章实现一个简化的 Raft 模拟器：5 个节点通过消息队列通信，模拟选举、心跳、日志复制的完整流程。

本章学习目标
  • 理解 Raft 的三种角色：Follower、Candidate、Leader
  • 理解 Term（任期）的作用和递增规则
  • 实现 Leader 选举（RequestVote RPC）
  • 实现心跳（AppendEntries RPC 空载荷）
  • 实现日志复制（AppendEntries 带日志条目）
  • 实现提交（commitIndex）和状态机应用

58.1 Raft 核心概念

// Raft 算法核心概念
//
// ┌─────────────────────────────────────────────────┐
// │  节点角色（三种，任何时刻每节点恰好一种）          │
// │  • Follower  — 被动响应 Leader/Candidate         │
// │  • Candidate — 发起选举，争取成为 Leader          │
// │  • Leader    — 处理客户端请求，复制日志           │
// ├─────────────────────────────────────────────────┤
// │  Term（任期）— 逻辑时钟，单调递增                 │
// │  • 每次选举开始新 Term                           │
// │  • 收到更高 Term → 立即退回 Follower             │
// ├─────────────────────────────────────────────────┤
// │  日志（Log）— 有序的操作记录                      │
// │  • 每条: { term, command }                      │
// │  • Leader 追加 → 复制给 Follower → 多数确认 → 提交│
// ├─────────────────────────────────────────────────┤
// │  两种 RPC：                                      │
// │  • RequestVote  — Candidate 请求投票             │
// │  • AppendEntries — Leader 发心跳/日志            │
// └─────────────────────────────────────────────────┘

Raft 的核心保证：任何时刻最多一个 Leader，且 Leader 拥有所有已提交的日志条目。

58.2 节点状态定义

// 节点角色常量
int FOLLOWER = 0
int CANDIDATE = 1
int LEADER = 2

class RaftNode
    int id                  // 节点 ID (0-4)
    int role                // 当前角色
    int currentTerm         // 当前任期
    int votedFor            // 本任期投给了谁 (-1=未投)
    int leaderId            // 当前 Leader (-1=未知)

    // 日志
    list<LogEntry> log      // 日志条目列表
    int commitIndex         // 已提交的最高索引
    int lastApplied         // 已应用的最高索引

    // Leader 专用
    list<int> nextIndex     // 每个 Follower 的下一条日志索引
    list<int> matchIndex    // 每个 Follower 已匹配的最高索引

    // 选举计时
    int electionTimeout     // 选举超时（随机 150-300ms）
    int lastHeartbeat       // 上次收到心跳的时间

    // 消息队列（模拟网络）
    list<Message> inbox     // 收件箱

    func init(int id, int nodeCount)
        this.id = id
        role = FOLLOWER
        currentTerm = 0
        votedFor = -1
        leaderId = -1
        log = list<LogEntry>()
        commitIndex = 0
        lastApplied = 0
        nextIndex = list<int>()
        matchIndex = list<int>()
        inbox = list<Message>()
        electionTimeout = 150 + math.random(0, 150)
        lastHeartbeat = time.now()

        int i = 0
        while i < nodeCount
            nextIndex.add(1)
            matchIndex.add(0)
            i += 1
        end
    end
end

每个节点维护自己的任期、日志、投票记录。nextIndex 和 matchIndex 只有 Leader 使用。

58.3 日志条目

// 日志条目
class LogEntry
    int term          // 写入时的任期
    string command    // 操作命令（如 "SET x 1"）

    func make(int term, string command) as LogEntry
        LogEntry e = LogEntry()
        e.term = term
        e.command = command
        return e
    end
end

// 消息（模拟 RPC）
class Message
    string type       // "RequestVote" / "VoteReply" / "AppendEntries" / "AppendReply"
    int from          // 发送者 ID
    int to            // 接收者 ID
    int term          // 发送者的任期

    // RequestVote 字段
    int lastLogIndex
    int lastLogTerm

    // AppendEntries 字段
    int prevLogIndex
    int prevLogTerm
    list<LogEntry> entries
    int leaderCommit

    // 回复字段
    bool success
    int matchIdx
end

// 辅助：获取节点最后一条日志的索引和任期
func lastLogInfo(RaftNode node) as array<int>
    array<int> result = array<int>(2)
    if node.log.len() == 0 then
        result[0] = 0   // index
        result[1] = 0   // term
    else
        result[0] = node.log.len()
        result[1] = node.log[node.log.len() - 1].term
    end
    return result
end

58.4 Leader 选举

当 Follower 超时未收到心跳，转为 Candidate 发起选举：

// 开始选举
func startElection(RaftNode node, list<RaftNode> cluster)
    node.role = CANDIDATE
    node.currentTerm += 1
    node.votedFor = node.id       // 投给自己
    node.leaderId = -1
    node.lastHeartbeat = time.now()
    node.electionTimeout = 150 + math.random(0, 150)  // 重新随机

    println("[Node " + node.id.toString() + "] 发起选举, Term=" + node.currentTerm.toString())

    // 获取自己的最后日志信息
    array<int> info = lastLogInfo(node)

    // 向所有其他节点发送 RequestVote
    int votes = 1   // 自己的一票
    int i = 0
    while i < cluster.len()
        if i != node.id then
            Message req = Message()
            req.type = "RequestVote"
            req.from = node.id
            req.to = i
            req.term = node.currentTerm
            req.lastLogIndex = info[0]
            req.lastLogTerm = info[1]

            // 处理投票
            Message reply = handleRequestVote(cluster[i], req)
            if reply.success then
                votes += 1
            end
        end
        i += 1
    end

    // 判断是否获得多数票
    int majority = cluster.len() / 2 + 1
    if votes >= majority and node.role == CANDIDATE then
        becomeLeader(node, cluster)
    end
end

// 成为 Leader
func becomeLeader(RaftNode node, list<RaftNode> cluster)
    node.role = LEADER
    node.leaderId = node.id
    println("[Node " + node.id.toString() + "] 当选 Leader! Term=" + node.currentTerm.toString())

    // 初始化 nextIndex 和 matchIndex
    int logLen = node.log.len() + 1
    int i = 0
    while i < cluster.len()
        node.nextIndex[i] = logLen
        node.matchIndex[i] = 0
        i += 1
    end

    // 立即发送心跳（空 AppendEntries）
    sendHeartbeats(node, cluster)
end

58.5 投票逻辑

节点收到 RequestVote 时的决策：

// 处理 RequestVote RPC
func handleRequestVote(RaftNode node, Message req) as Message
    Message reply = Message()
    reply.type = "VoteReply"
    reply.from = node.id
    reply.to = req.from
    reply.term = node.currentTerm
    reply.success = false

    // 规则 1：如果请求的 Term 比自己的大，更新自己
    if req.term > node.currentTerm then
        node.currentTerm = req.term
        node.role = FOLLOWER
        node.votedFor = -1
        reply.term = req.term
    end

    // 规则 2：Term 不匹配 → 拒绝
    if req.term < node.currentTerm then
        return reply
    end

    // 规则 3：本任期已投给别人 → 拒绝
    if node.votedFor != -1 and node.votedFor != req.from then
        return reply
    end

    // 规则 4：候选人的日志必须至少和自己一样新
    array<int> myInfo = lastLogInfo(node)
    int myLastTerm = myInfo[1]
    int myLastIndex = myInfo[0]

    bool logOk = false
    if req.lastLogTerm > myLastTerm then
        logOk = true
    elif req.lastLogTerm == myLastTerm and req.lastLogIndex >= myLastIndex then
        logOk = true
    end

    if logOk then
        node.votedFor = req.from
        reply.success = true
        println("  [Node " + node.id.toString() + "] 投票给 Node " + req.from.toString())
    end

    return reply
end

投票的四个条件：① Term 有效；② 本任期未投票；③ 候选人日志足够新。这保证了只有拥有最新日志的节点才能当选。

58.6 心跳机制

Leader 定期发送空 AppendEntries（心跳）维持权威：

// Leader 发送心跳给所有 Follower
func sendHeartbeats(RaftNode node, list<RaftNode> cluster)
    int i = 0
    while i < cluster.len()
        if i != node.id then
            Message hb = Message()
            hb.type = "AppendEntries"
            hb.from = node.id
            hb.to = i
            hb.term = node.currentTerm
            hb.prevLogIndex = node.nextIndex[i] - 1
            hb.prevLogTerm = 0
            hb.entries = list<LogEntry>()   // 空 = 心跳
            hb.leaderCommit = node.commitIndex

            // 获取 prevLogTerm
            if hb.prevLogIndex > 0 and hb.prevLogIndex <= node.log.len() then
                hb.prevLogTerm = node.log[hb.prevLogIndex - 1].term
            end

            // 发送并处理回复
            Message reply = handleAppendEntries(cluster[i], hb)
            processAppendReply(node, i, reply)
        end
        i += 1
    end
end

// Follower 处理心跳/日志
func handleAppendEntries(RaftNode node, Message req) as Message
    Message reply = Message()
    reply.type = "AppendReply"
    reply.from = node.id
    reply.to = req.from
    reply.term = node.currentTerm
    reply.success = false

    // Term 检查
    if req.term < node.currentTerm then
        return reply   // 过期 Leader，拒绝
    end

    // 认可 Leader
    if req.term > node.currentTerm then
        node.currentTerm = req.term
    end
    node.role = FOLLOWER
    node.leaderId = req.from
    node.votedFor = -1
    node.lastHeartbeat = time.now()   // 重置选举计时器
    reply.term = node.currentTerm

    // 日志一致性检查
    if req.prevLogIndex > 0 then
        if req.prevLogIndex > node.log.len() then
            return reply   // 日志不够长，拒绝
        end
        if node.log[req.prevLogIndex - 1].term != req.prevLogTerm then
            // 日志冲突，删除冲突条目
            while node.log.len() > req.prevLogIndex - 1
                node.log.removeAt(node.log.len() - 1)
            end
            return reply
        end
    end

    // 追加新条目
    int i = 0
    while i < req.entries.len()
        int idx = req.prevLogIndex + i + 1
        if idx <= node.log.len() then
            // 已存在：检查冲突
            if node.log[idx - 1].term != req.entries[i].term then
                // 冲突：删除此位置及之后所有
                while node.log.len() > idx - 1
                    node.log.removeAt(node.log.len() - 1)
                end
                node.log.add(req.entries[i])
            end
        else
            // 不存在：追加
            node.log.add(req.entries[i])
        end
        i += 1
    end

    // 更新 commitIndex
    if req.leaderCommit > node.commitIndex then
        int newCommit = req.leaderCommit
        if newCommit > node.log.len() then
            newCommit = node.log.len()
        end
        node.commitIndex = newCommit
    end

    reply.success = true
    reply.matchIdx = node.log.len()
    return reply
end

58.7 日志复制

客户端请求到达 Leader 后的处理流程：

// Leader 处理客户端命令
func leaderAppend(RaftNode node, list<RaftNode> cluster, string command)
    if node.role != LEADER then
        println("  不是 Leader，请求转发给 Node " + node.leaderId.toString())
        return
    end

    // 1. 追加到本地日志
    LogEntry entry = LogEntry.make(node.currentTerm, command)
    node.log.add(entry)
    println("[Leader " + node.id.toString() + "] 追加日志: [" +
            node.currentTerm.toString() + "] " + command)

    // 2. 复制给所有 Follower
    int i = 0
    while i < cluster.len()
        if i != node.id then
            Message req = Message()
            req.type = "AppendEntries"
            req.from = node.id
            req.to = i
            req.term = node.currentTerm
            req.prevLogIndex = node.nextIndex[i] - 1
            req.prevLogTerm = 0
            req.entries = list<LogEntry>()
            req.entries.add(entry)
            req.leaderCommit = node.commitIndex

            if req.prevLogIndex > 0 and req.prevLogIndex <= node.log.len() then
                req.prevLogTerm = node.log[req.prevLogIndex - 1].term
            end

            Message reply = handleAppendEntries(cluster[i], req)
            processAppendReply(node, i, reply)
        end
        i += 1
    end

    // 3. 检查是否可以提交
    advanceCommitIndex(node, cluster)
end

// 处理 AppendEntries 回复
func processAppendReply(RaftNode node, int peerId, Message reply)
    if reply.term > node.currentTerm then
        // 发现更高 Term，退回 Follower
        node.currentTerm = reply.term
        node.role = FOLLOWER
        node.votedFor = -1
        return
    end

    if reply.success then
        node.nextIndex[peerId] = reply.matchIdx + 1
        node.matchIndex[peerId] = reply.matchIdx
    else
        // 失败：回退 nextIndex 重试
        if node.nextIndex[peerId] > 1 then
            node.nextIndex[peerId] -= 1
        end
    end
end

58.8 提交与应用

// 推进 commitIndex：找到多数节点已复制的最高索引
func advanceCommitIndex(RaftNode node, list<RaftNode> cluster)
    int n = node.log.len()
    while n > node.commitIndex
        // 统计有多少节点复制到了索引 n
        int count = 1   // Leader 自己
        int i = 0
        while i < cluster.len()
            if i != node.id and node.matchIndex[i] >= n then
                count += 1
            end
            i += 1
        end

        // 多数确认 + 必须是当前 Term 的日志
        int majority = cluster.len() / 2 + 1
        if count >= majority and node.log[n - 1].term == node.currentTerm then
            node.commitIndex = n
            println("[Leader " + node.id.toString() + "] 提交索引 " + n.toString())
            break
        end
        n -= 1
    end

    // 应用已提交的日志到状态机
    applyCommitted(node)
end

// 应用已提交但未应用的日志
func applyCommitted(RaftNode node)
    while node.lastApplied < node.commitIndex
        node.lastApplied += 1
        LogEntry entry = node.log[node.lastApplied - 1]
        println("  [Node " + node.id.toString() + "] 应用: " + entry.command)
    end
end

提交条件：① 多数节点（含 Leader）已复制；② 该条目属于当前 Term。提交后，所有节点最终都会应用相同的命令序列——这就是共识。

58.9 模拟网络

本章用同步调用模拟 RPC（教学清晰）。生产环境用 TCP/UDP：

// 模拟网络层
// 真实实现中，每个节点是一个线程/进程，通过 TCP 通信
// 本章简化为：直接函数调用（同步 RPC）
//
// 生产架构：
// ┌────────┐  TCP   ┌────────┐  TCP   ┌────────┐
// │ Node 0 │◄─────►│ Node 1 │◄─────►│ Node 2 │
// │ Leader │        │Follower│        │Follower│
// └────────┘        └────────┘        └────────┘
//      ▲                                  ▲
//      │ TCP                         TCP  │
//      ▼                                  ▼
// ┌────────┐                         ┌────────┐
// │ Node 3 │                         │ Node 4 │
// │Follower│                         │Follower│
// └────────┘                         └────────┘
//
// 模拟 vs 真实的区别：
// • 模拟：同步调用，无网络延迟，无丢包
// • 真实：异步 + 超时 + 重试 + 乱序
// • 核心逻辑完全相同！

// 模拟选举超时检查
func checkElectionTimeout(RaftNode node, list<RaftNode> cluster)
    if node.role == LEADER then return end

    int elapsed = time.now() - node.lastHeartbeat
    if elapsed > node.electionTimeout then
        println("[Node " + node.id.toString() + "] 选举超时 (" +
                elapsed.toString() + "ms)")
        startElection(node, cluster)
    end
end
为什么用同步模拟

Raft 的难点在算法逻辑（选举规则、日志一致性），不在网络传输。同步模拟去掉了网络不确定性，让你专注于理解算法本身。掌握后，将函数调用替换为 TCP send/recv 就是真实的分布式 Raft。

58.10 完整模拟

组装所有组件，运行一次完整的 Raft 流程：

// ═══════════════════════════════════════
//  raft_sim.xl — Raft 共识模拟
// ═══════════════════════════════════════

func main()
    int NODE_COUNT = 5
    list<RaftNode> cluster = list<RaftNode>()

    // 初始化 5 个节点
    int i = 0
    while i < NODE_COUNT
        RaftNode node = RaftNode()
        node.init(i, NODE_COUNT)
        cluster.add(node)
        i += 1
    end
    println("═══ Raft 集群初始化: " + NODE_COUNT.toString() + " 个节点 ═══")
    println("")

    // ─── 阶段 1：选举 ───
    println("─── 阶段 1: Leader 选举 ───")
    // 模拟 Node 0 超时，发起选举
    startElection(cluster[0], cluster)
    println("")

    // ─── 阶段 2：心跳 ───
    println("─── 阶段 2: Leader 心跳 ───")
    RaftNode leader = cluster[0]
    sendHeartbeats(leader, cluster)
    println("所有 Follower 收到心跳，选举计时器重置")
    println("")

    // ─── 阶段 3：日志复制 ───
    println("─── 阶段 3: 日志复制 ───")
    leaderAppend(leader, cluster, "SET x 1")
    leaderAppend(leader, cluster, "SET y 2")
    leaderAppend(leader, cluster, "SET x 10")
    println("")

    // ─── 阶段 4：验证状态 ───
    println("─── 阶段 4: 集群状态 ───")
    i = 0
    while i < NODE_COUNT
        RaftNode n = cluster[i]
        string roleName = "Follower"
        if n.role == CANDIDATE then roleName = "Candidate" end
        if n.role == LEADER then roleName = "Leader" end

        println("Node " + n.id.toString() + ": " + roleName +
                " | Term=" + n.currentTerm.toString() +
                " | Log=" + n.log.len().toString() +
                " | Commit=" + n.commitIndex.toString() +
                " | Applied=" + n.lastApplied.toString())
        i += 1
    end
    println("")

    // ─── 阶段 5：模拟 Leader 故障 + 重新选举 ───
    println("─── 阶段 5: Leader 故障, 重新选举 ───")
    println("[Node 0] 宕机!")
    // Node 1 超时，发起新选举
    cluster[1].lastHeartbeat = time.now() - 500  // 模拟超时
    checkElectionTimeout(cluster[1], cluster)
    println("")

    // 新 Leader 继续服务
    println("─── 阶段 6: 新 Leader 继续服务 ───")
    RaftNode newLeader = cluster[1]
    leaderAppend(newLeader, cluster, "SET z 99")
    println("")

    // 最终状态
    println("═══ 最终状态 ═══")
    i = 0
    while i < NODE_COUNT
        RaftNode n = cluster[i]
        string roleName = "Follower"
        if n.role == CANDIDATE then roleName = "Candidate" end
        if n.role == LEADER then roleName = "Leader" end
        println("Node " + n.id.toString() + ": " + roleName +
                " | Term=" + n.currentTerm.toString() +
                " | Log=" + n.log.len().toString() +
                " | Commit=" + n.commitIndex.toString())
        i += 1
    end
end

main()

预期输出（简化）：

═══ Raft 集群初始化: 5 个节点 ═══

─── 阶段 1: Leader 选举 ───
[Node 0] 发起选举, Term=1
  [Node 1] 投票给 Node 0
  [Node 2] 投票给 Node 0
  [Node 3] 投票给 Node 0
  [Node 4] 投票给 Node 0
[Node 0] 当选 Leader! Term=1

─── 阶段 3: 日志复制 ───
[Leader 0] 追加日志: [1] SET x 1
[Leader 0] 提交索引 1
  [Node 0] 应用: SET x 1
  [Node 1] 应用: SET x 1
  ...

─── 阶段 5: Leader 故障, 重新选举 ───
[Node 0] 宕机!
[Node 1] 选举超时 (500ms)
[Node 1] 发起选举, Term=2
[Node 1] 当选 Leader! Term=2

深入细节

1. 为什么选举超时是随机的？

如果所有节点超时相同，会同时发起选举，互相投票导致"分裂投票"（没人获得多数）。随机超时（150-300ms）使某个节点大概率率先超时、率先竞选、率先获得多数票。

2. 日志一致性如何保证？

AppendEntries 携带 prevLogIndex + prevLogTerm。Follower 检查该位置的日志是否匹配——不匹配则拒绝，Leader 回退 nextIndex 重试。这保证了日志匹配属性：如果两个节点在相同索引有相同 Term，则该索引及之前的所有日志都相同。

3. 为什么只能提交当前 Term 的日志？

Raft 论文 Figure 8 描述了一个场景：Leader 不能通过计数方式直接提交前任 Term 的日志。必须通过提交当前 Term 的新条目，间接提交之前的旧条目。这避免了已提交日志被覆盖。

4. 从模拟到生产

将本章的同步调用替换为：① TCP 连接 + 消息序列化（第 45 章）；② 每节点一个线程（第 57 章）；③ 定时器驱动心跳和选举超时；④ 持久化日志（第 54 章）。你就得到了一个真实的 Raft 实现。

⚠️ 常见错误

错误 1：收到高 Term 不退回 Follower

// ✗ 忽略 Term 检查
func handleMsg(RaftNode node, Message msg)
    // 直接处理...
end

// ✓ 任何 RPC 回复中 Term > currentTerm → 立即退回
if msg.term > node.currentTerm then
    node.currentTerm = msg.term
    node.role = FOLLOWER
    node.votedFor = -1
end

错误 2：Candidate 赢得选举后不重置 nextIndex

// ✗ nextIndex 残留旧值，日志复制从错误位置开始
// ✓ 当选后立即初始化 nextIndex[i] = 本地日志长度 + 1

错误 3：日志索引从 0 还是 1 开始混乱

// Raft 论文中日志索引从 1 开始
// log[0] 对应索引 1，log[len-1] 对应索引 len
// prevLogIndex=0 表示"之前没有日志"（哨兵）
// 务必统一！本章：索引 1-based，数组 0-based

错误 4：Follower 接受过期 Leader 的日志

// ✗ 不检查 Term 就追加
// ✓ AppendEntries 的 Term < currentTerm → 拒绝（return success=false）

本章小结

  • 三种角色：Follower（被动）→ Candidate（选举）→ Leader（服务）
  • Term：逻辑时钟，单调递增，用于识别过期信息
  • 选举：超时 → 自增 Term → 请求投票 → 多数当选
  • 心跳：Leader 定期发空 AppendEntries，阻止 Follower 超时
  • 日志复制：Leader 追加 → 复制 → 多数确认 → 提交 → 应用
  • 安全性：日志匹配属性 + 只提交当前 Term + 选举限制
  • 全书收官：从 Hello World 到分布式共识，你已走完 xlang 的完整旅程
恭喜完成全书！

从第 1 章的 Hello World 到第 58 章的 Raft 共识，你经历了变量、控制流、函数、面向对象、泛型、并发、网络、文件 I/O，最终到达分布式系统。这不仅是 xlang 的旅程，更是计算机科学核心知识的完整脉络。继续实践，继续构建！

复习题

  1. Raft 为什么需要随机选举超时？如果所有节点用相同超时会怎样？
  2. 一个节点在什么条件下会给 Candidate 投票？列出所有条件。
  3. Leader 如何判断一条日志可以"提交"？为什么不能提交旧 Term 的日志？
  4. AppendEntries 的 prevLogIndex/prevLogTerm 起什么作用？不匹配时如何处理？
  5. 如果 5 节点集群中 3 个宕机，能否选出 Leader？为什么？

编程练习

练习 1：持久化日志

将 Raft 日志持久化到文件（参考第 54 章 KV 存储）。节点重启后从文件恢复日志和 currentTerm。提示：每条日志一行 term\tcommand。

练习 2：真实网络 Raft

将模拟 RPC 替换为 TCP 通信（参考第 45、57 章）。每个节点是一个线程，监听自己的端口。消息用 JSON 序列化（参考第 52 章）。提示：消息格式 {"type":"RequestVote","term":1,...}\n。

练习 3：日志压缩（快照）

当日志超过 100 条时，生成快照（保存状态机 + 最后日志索引），截断旧日志。新节点加入时直接发送快照而非全部日志。提示：快照 = commitIndex + 状态 dict 的序列化。