Raft 共识算法学习笔记

Raft 是过去十年影响力最大的共识算法：Etcd、Consul、TiKV、Nacos 都以它为复制内核。Ousterhout 的设计目标非常明确——先保证可理解性，再追求与 Paxos 等价的安全性。本文给出一个约 400 行、完整可运行的 Go 实现：三种角色的状态转换、随机超时选举、心跳维持、日志复制与一致性检查、提交推进，并用分区模拟验证 Raft 的两条核心安全性：选举安全（一任期至多一个 Leader）与日志匹配（已提交日志永不丢失）。

一、Raft 的问题分解

Raft 把共识拆成三个相对独立的子问题，这是它比 Paxos 易懂的根源：

• Leader 选举：集群任意时刻至多一个 Leader；Leader 宕机后，剩余节点用随机超时竞选出继任者。
• 日志复制：所有客户端写都经 Leader 接收、复制到多数派后才提交；Follower 被动接受，日志冲突时以 Leader 为准强制覆盖。
• 安全性约束：通过投票限制（日志不比候选者旧才投票）和提交限制（Leader 只能提交当前任期的日志）两条规则，保证已提交日志在任何故障下都不丢失。

节点状态机

每个节点在三种角色间流转：Follower（被动响应）、Candidate（选举中）、Leader（服务写请求）。转换条件：Follower 超时未收到心跳 → 变 Candidate 发起选举；Candidate 获多数票 → 变 Leader；任何角色发现更高任期 → 退回 Follower 并更新任期。

二、完整实现

2.1 核心数据结构

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type Role int

const (
    Follower Role = iota
    Candidate
    Leader
)

func (r Role) String() string {
    return []string{"Follower", "Candidate", "Leader"}[r]
}

// LogEntry 单条日志项
type LogEntry struct {
    Term    int    // 创建该日志的 Leader 任期
    Command string // 状态机命令
}

// RaftNode 一个 Raft 节点。注释标明论文中的持久化/易失状态
type RaftNode struct {
    mu   sync.Mutex
    id   int
    role Role

    // ---- 持久化状态（宕机重启必须保留）----
    currentTerm int        // 见过的最大任期
    votedFor    int        // 本任期投给谁，-1 表示未投
    log         []LogEntry // 日志，索引从 0 开始

    // ---- 易失状态 ----
    commitIndex int // 已提交的最大日志索引
    lastApplied int // 已应用到状态机的最大索引

    // ---- Leader 专有状态 ----
    nextIndex  []int // 每个 Follower 下一条待复制日志的索引
    matchIndex []int // 每个 Follower 已确认复制的最大索引

    peers      []*RaftNode
    hearbeatCh chan bool  // 收到心跳/授权投票的信号
    applyCh    chan LogEntry
    shutdown   chan struct{}
}

2.2 选举计时器：随机超时的实现

随机化选举超时是 Raft 活性的关键：所有节点超时不同步，总有一个先醒来发起选举并胜出，几乎不会出现选票持续分裂：

func NewRaftNode(id int) *RaftNode {
    n := &RaftNode{
        id: id, role: Follower, votedFor: -1,
        hearbeatCh: make(chan bool, 1),
        applyCh:    make(chan LogEntry, 64),
        shutdown:   make(chan struct{}),
    }
    go n.electionLoop()
    return n
}

// electionTimeout 随机化选举超时 [150ms, 300ms)
func electionTimeout() time.Duration {
    return time.Duration(150+rand.Intn(150)) * time.Millisecond
}

// electionLoop Follower/Candidate 的选举计时循环
func (n *RaftNode) electionLoop() {
    for {
        select {
        case <-n.shutdown:
            return
        case <-time.After(electionTimeout()):
            n.mu.Lock()
            role := n.role
            n.mu.Unlock()
            if role != Leader {
                n.startElection() // 超时未收到心跳 → 发起选举
            }
        case <-n.hearbeatCh:
            // 收到有效心跳或投了票，重置计时器
        }
    }
}

2.3 发起选举与 RequestVote 处理

// startElection 候选人流程：自增任期、投自己、并发拉票
func (n *RaftNode) startElection() {
    n.mu.Lock()
    n.role = Candidate
    n.currentTerm++
    term := n.currentTerm
    n.votedFor = n.id
    lastIdx := len(n.log) - 1
    lastTerm := 0
    if lastIdx >= 0 {
        lastTerm = n.log[lastIdx].Term
    }
    n.mu.Unlock()

    fmt.Printf("[N%d] 任期 %d: 发起选举 (lastLog=%d@%d)\n",
        n.id, term, lastIdx, lastTerm)
    votes := 1 // 先投自己
    var vmu sync.Mutex
    var wg sync.WaitGroup

    for _, p := range n.peers {
        if p.id == n.id {
            continue
        }
        wg.Add(1)
        go func(peer *RaftNode) {
            defer wg.Done()
            if peer.requestVote(term, n.id, lastIdx, lastTerm) {
                vmu.Lock()
                votes++
                vmu.Unlock()
            }
        }(p)
    }
    wg.Wait()

    n.mu.Lock()
    defer n.mu.Unlock()
    if n.role == Candidate && n.currentTerm == term {
        if votes >= len(n.peers)/2+1 {
            n.becomeLeader()
        } else {
            fmt.Printf("[N%d] 任期 %d: 仅获 %d 票，选举失败\n",
                n.id, term, votes)
        }
    }
}

// requestVote RequestVote RPC 的处理端（在接收方节点上执行）
func (n *RaftNode) requestVote(term, candID, candLastIdx, candLastTerm int) bool {
    n.mu.Lock()
    defer n.mu.Unlock()
    if term < n.currentTerm {
        return false // 拒绝旧任期
    }
    if term > n.currentTerm {
        n.currentTerm = term
        n.role = Follower
        n.votedFor = -1
    }
    // 日志新旧比较：先比最后一条的任期，再比长度
    myLastIdx := len(n.log) - 1
    myLastTerm := 0
    if myLastIdx >= 0 {
        myLastTerm = n.log[myLastIdx].Term
    }
    upToDate := candLastTerm > myLastTerm ||
        (candLastTerm == myLastTerm && candLastIdx >= myLastIdx)
    if (n.votedFor == -1 || n.votedFor == candID) && upToDate {
        n.votedFor = candID
        select {
        case n.hearbeatCh <- true: // 重置选举计时
        default:
        }
        return true
    }
    return false
}

「日志新旧比较」是选举安全的核心：一个日志落后于多数派的节点永远拿不到多数票，因此已提交的日志必然出现在未来每个 Leader 身上。

2.4 Leader 上任与日志复制

func (n *RaftNode) becomeLeader() {
    n.role = Leader
    n.nextIndex = make([]int, len(n.peers))
    n.matchIndex = make([]int, len(n.peers))
    for i := range n.peers {
        n.nextIndex[i] = len(n.log) // 乐观假设：Follower 只差 0 条
        n.matchIndex[i] = 0
    }
    fmt.Printf("[N%d] 任期 %d: 当选 Leader (获多数票)\n", n.id, n.currentTerm)
    go n.heartbeatLoop()
}

// heartbeatLoop 周期心跳：空的 AppendEntries 兼作保活与提交推进
func (n *RaftNode) heartbeatLoop() {
    ticker := time.NewTicker(50 * time.Millisecond) // 心跳 << 选举超时
    defer ticker.Stop()
    for {
        n.mu.Lock()
        if n.role != Leader {
            n.mu.Unlock()
            return
        }
        n.mu.Unlock()
        select {
        case <-n.shutdown:
            return
        case <-ticker.C:
            for _, p := range n.peers {
                if p.id != n.id {
                    go n.replicateTo(p)
                }
            }
        }
    }
}

// replicateTo 向单个 Follower 复制日志（含一致性检查与回退）
func (n *RaftNode) replicateTo(p *RaftNode) {
    n.mu.Lock()
    next := n.nextIndex[p.id]
    prevIdx := next - 1
    prevTerm := 0
    if prevIdx >= 0 && prevIdx < len(n.log) {
        prevTerm = n.log[prevIdx].Term
    }
    entries := append([]LogEntry{}, n.log[next:]...)
    term := n.currentTerm
    leaderCommit := n.commitIndex
    n.mu.Unlock()

    ok, conflict := p.appendEntries(term, n.id, prevIdx, prevTerm,
        entries, leaderCommit)
    n.mu.Lock()
    defer n.mu.Unlock()
    if n.role != Leader || n.currentTerm != term {
        return
    }
    if ok {
        n.matchIndex[p.id] = prevIdx + len(entries)
        n.nextIndex[p.id] = n.matchIndex[p.id] + 1
        n.advanceCommit() // 尝试推进提交点
    } else if conflict {
        n.nextIndex[p.id]-- // 回退一格，下次心跳重试
    }
}

2.5 AppendEntries：一致性检查

// appendEntries AppendEntries RPC 的处理端
func (n *RaftNode) appendEntries(term, leaderID, prevIdx, prevTerm int,
    entries []LogEntry, leaderCommit int) (bool, bool) {
    n.mu.Lock()
    defer n.mu.Unlock()
    if term < n.currentTerm {
        return false, false // 拒绝旧任期 Leader
    }
    if term >= n.currentTerm {
        n.currentTerm = term
        n.role = Follower
        n.votedFor = -1
    }
    select {
    case n.hearbeatCh <- true: // 有效 Leader 心跳，重置选举计时
    default:
    }

    // 一致性检查：prevIdx 位置的日志任期必须匹配
    if prevIdx >= 0 {
        if prevIdx >= len(n.log) || n.log[prevIdx].Term != prevTerm {
            return false, true // 冲突：通知 Leader 回退 nextIndex
        }
    }
    // 追加：与已有日志冲突的部分直接截断覆盖（Leader 权威）
    for i, e := range entries {
        idx := prevIdx + 1 + i
        if idx < len(n.log) {
            if n.log[idx].Term != e.Term {
                n.log = n.log[:idx] // 删除冲突点及之后全部
            }
        }
        if idx >= len(n.log) {
            n.log = append(n.log, e)
        }
    }
    // 推进本地提交点
    if leaderCommit > n.commitIndex {
        n.commitIndex = min(leaderCommit, len(n.log)-1)
        n.applyCommitted()
    }
    return true, false
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

2.6 提交推进与安全限制

// advanceCommit Leader 推进提交点：只提交当前任期的日志
type RaftNode2 = RaftNode

func (n *RaftNode) advanceCommit() {
    for idx := n.commitIndex + 1; idx < len(n.log); idx++ {
        // 关键安全规则：只能按任期提交——
        // 旧任期的日志即使复制到多数派，也不能直接提交，
        // 必须等当前任期至少一条日志落多数派后顺带提交。
        if n.log[idx].Term != n.currentTerm {
            continue
        }
        count := 1 // Leader 自己
        for i := range n.peers {
            if i != n.id && n.matchIndex[i] >= idx {
                count++
            }
        }
        if count >= len(n.peers)/2+1 {
            n.commitIndex = idx
        }
    }
    n.applyCommitted()
}

// applyCommitted 把已提交日志按序应用到状态机
func (n *RaftNode) applyCommitted() {
    for n.lastApplied < n.commitIndex {
        n.lastApplied++
        select {
        case n.applyCh <- n.log[n.lastApplied]:
        default:
        }
    }
}

// ClientWrite 客户端写入口：仅 Leader 可处理
func (n *RaftNode) ClientWrite(cmd string) bool {
    n.mu.Lock()
    defer n.mu.Unlock()
    if n.role != Leader {
        return false
    }
    n.log = append(n.log, LogEntry{Term: n.currentTerm, Command: cmd})
    fmt.Printf("[N%d] 任期 %d: 接收客户端写入 '%s' (日志 #%d)\n",
        n.id, n.currentTerm, cmd, len(n.log)-1)
    for _, p := range n.peers {
        if p.id != n.id {
            go n.replicateTo(p)
        }
    }
    return true
}

advanceCommit 中的任期检查对应论文 Figure 8 的经典反例：旧任期日志被复制到多数派后，原 Leader 宕机，新 Leader 可能覆盖它——因此 Raft 只允许直接提交当前任期的日志，旧日志靠当前任期日志的提交「顺带」固化。

三、模拟演示

3.1 五节点集群的选举与复制

func main() {
    rand.Seed(time.Now().UnixNano())
    nodes := make([]*RaftNode, 5)
    for i := range nodes {
        nodes[i] = NewRaftNode(i)
    }
    for i := range nodes {
        nodes[i].peers = nodes
    }
    time.Sleep(500 * time.Millisecond) // 等待选举完成

    // 找到 Leader 并写入
    for _, n := range nodes {
        if n.ClientWrite("SET x = 1") {
            break
        }
    }
    time.Sleep(300 * time.Millisecond) // 等待复制与提交

    for _, n := range nodes {
        n.mu.Lock()
        fmt.Printf("N%d: role=%s term=%d log=%v commit=%d\n",
            n.id, n.role, n.currentTerm, n.log, n.commitIndex)
        n.mu.Unlock()
    }
}

// 典型输出:
// [N2] 任期 1: 发起选举 (lastLog=-1@0)
// [N2] 任期 1: 当选 Leader (获多数票)
// [N2] 任期 1: 接收客户端写入 'SET x = 1' (日志 #0)
// N0: role=Follower term=1 log=[{1 SET x = 1}] commit=0
// N1: role=Follower term=1 log=[{1 SET x = 1}] commit=0
// N2: role=Leader   term=1 log=[{1 SET x = 1}] commit=0
// N3: role=Follower term=1 log=[{1 SET x = 1}] commit=0
// N4: role=Follower term=1 log=[{1 SET x = 1}] commit=0

3.2 网络分区：少数派无法选主，多数派照常服务

// 把 {N0,N1} 与 {N2,N3,N4} 断开（peers 列表改为只含同分区节点）:
// - 少数派 {N0,N1}: 反复发起选举，任期不断膨胀，但永远集不齐 3 票，
//   不会选出第二个 Leader —— 选举安全成立。
// - 多数派 {N2,N3,N4}: Leader 正常心跳、正常提交新日志。
// - 分区恢复后: 少数派节点发现更高任期的心跳，退回 Follower，
//   通过 nextIndex 回退机制补齐缺失的日志，集群重新一致。

3.3 脑裂不会发生的形式化直觉

假设任期 T 有两个 Leader L1、L2。L1 获多数派 M1 投票，L2 获多数派 M2 投票。由鸽巢原理 M1∩M2 非空，其中必有节点对同一任期投了两票——与 votedFor 每任期只投一次矛盾。因此选举安全由「每任期一票」直接推出，与网络时序无关。这也是 Raft 不需要时钟同步的原因：正确性全部建立在任期与索引的偏序比较上。

四、与 Paxos 的对照

五、生产实现的关键优化（对照 Etcd）

• 批量与流水线：Etcd 把多个客户端写打包进一次 AppendEntries，并用 pipeline 异步刷盘与网络发送，吞吐提升 5~10 倍。
• PreVote：节点发起选举前先做一次「预投票」，确认自己能拿到多数票才真正自增任期——避免分区节点恢复后用膨胀的任期打翻现任 Leader。
• Learner 角色：新加入节点先以 Learner 身份只接收日志不参与投票，追平后再转正，避免扩容瞬间多数派抖动。
• 快照压缩：日志超过阈值（Etcd 默认 10000 条）打快照并截断，落后太多的 Follower 直接收快照。
• ReadIndex：线性一致读不走日志：Leader 确认自己仍是多数派承认的 Leader 后，按当前 commitIndex 应答，省去一次写盘。

六、常见误区

• 误区一：心跳只是保活。心跳还携带 leaderCommit 推进 Follower 的提交点，并隐含「当前任期 Leader 存活」的授权。
• 误区二：日志复制到多数派就等于已提交。缺少任期检查会踩中 Figure 8 反例，旧任期日志可能被覆盖。
• 误区三：选举超时固定即可。固定超时必然出现选票无限分裂；随机化是活性的唯一保障。
• 误区四：Follower 拒绝不一致日志是异常。这是正常的一致性检查流程，Leader 通过 nextIndex 回退最终一定能找到匹配点。

七、总结

本文的 400 行实现覆盖了 Raft 论文的核心：选举计时器、RequestVote 的日志新鲜度检查、AppendEntries 的前缀匹配与冲突截断、基于任期的提交限制。Raft 的工程之美在于它把安全性压缩进少数几条局部规则——每任期一票、日志新鲜才投票、只提交当前任期——每条都能在几十行代码内说清。理解了这三条，再看 Etcd 的源码，你会发现剩下的都是性能优化与运维细节。

参考文献

1. D. Ongaro, J. Ousterhout. In Search of an Understandable Consensus Algorithm (Extended Version). USENIX ATC, 2014.
2. D. Ongaro. Consensus: Bridging Theory and Practice. Stanford PhD Thesis, 2014.
3. Etcd 官方文档与 raft 库源码: github.com/etcd-io/raft.
4. The Raft Consensus Algorithm 可视化: raft.github.io.