Raft 一致性协议：评测集群选主与日志复制的核心机制

Raft 一致性协议评测集群选主与日志复制的核心机制一、所有 Worker 都说自己是 Leader上上周半夜评测集群的 Leader 节点突然 GC 暂停了 5 秒。集群内的其他节点检测到心跳超时立即发起选举。但 Leader 恢复后又继续发心跳——此时集群里出现了两个Leader。提交任务被随机分配到两个节点状态完全错乱不得不全量重启。这不是 Raft 协议的 bug而是一次经典的**脑裂split-brain**场景。Raft 通过 term任期机制来处理这种情况每个选举周期都有一个递增的 term 编号节点只接受当前 term 或更高 term 的 Leader。两个 Leader 不可能同时赢得同一个 term 的选举。Raft 不是 Paxos 的简化版——它是一种工程上可理解的选择而共识达成协议。核心思想是Leader-based 复制所有写操作先到 LeaderLeader 将日志复制到多数 Follower 后提交最后应用到状态机。二、Raft 核心选举、日志复制、安全性三个子问题Leader 选举集群没有 Leader 时节点发起投票先获得多数票的成为新 Leader日志复制Leader 接收客户端请求转换为日志条目复制到所有 Follower安全性保证已提交的日志条目不会丢失import time import random import threading from enum import Enum from typing import List, Dict, Optional, Tuple from dataclasses import dataclass, field class NodeState(Enum): FOLLOWER follower CANDIDATE candidate LEADER leader dataclass class LogEntry: Raft 日志条目 每个条目包含 - term: 此条目被创建时的 Leader 任期 - index: 日志中的位置索引从 1 开始 - command: 要应用到状态机的命令 term: int index: int command: str dataclass class RaftConfig: Raft 节点配置 node_id: str peers: List[str] # 集群中其他节点的 ID election_timeout_min: int 150 # 最小选举超时ms election_timeout_max: int 300 # 最大选举超时ms heartbeat_interval: int 50 # 心跳间隔ms class RaftNode: Raft 共识协议的简化实现 实现了 Leader 选举和日志复制省略了 - 快照snapshotting - 成员变更membership changes - 网络分区处理生产环境需要 三个核心 RPC实际由网络通信实现这里用方法调用模拟 - RequestVote: Candidate → 所有节点请求投票 - AppendEntries: Leader → Followers心跳 日志复制 def __init__(self, config: RaftConfig): self.config config self.node_id config.node_id self.peers config.peers # 持久化状态所有节点 self.current_term: int 0 self.voted_for: Optional[str] None self.log: List[LogEntry] [] # 易失性状态所有节点 self.commit_index: int 0 self.last_applied: int 0 # 易失性状态仅 Leader self.next_index: Dict[str, int] {} # 每个 Follower 的下一个日志索引 self.match_index: Dict[str, int] {} # 每个 Follower 已复制的最高索引 # 运行时状态 self.state NodeState.FOLLOWER self.leader_id: Optional[str] None self.election_timer self._reset_election_timer() self._lock threading.Lock() self._state_machine: Dict[str, str] {} # 简化的键值状态机 def _reset_election_timer(self) - float: 重置选举超时计时器 随机超时是 Raft 的核心设计 如果所有节点同时超时它们会同时发起选举 选票分散导致无人过半。随机超时保证了 不同节点在不同时间发起选举。 timeout_ms random.uniform( self.config.election_timeout_min, self.config.election_timeout_max ) return time.time() timeout_ms / 1000.0 def start_election(self): 发起 Leader 选举 步骤 1. 增加 term 2. 投票给自己 3. 向所有 peers 发送 RequestVote 4. 统计选票 - 获得多数 → 成为 Leader - 收到更高 term → 退为 Follower - 超时未过半 → 重新选举 with self._lock: self.current_term 1 self.state NodeState.CANDIDATE self.voted_for self.node_id # 先投票给自己 self.leader_id None print(f[{self.node_id}] 发起选举 term{self.current_term}) votes 1 # 自己的一票 total_nodes len(self.config.peers) 1 # peers self majority total_nodes // 2 1 for peer_id in self.config.peers: # 模拟向每个 peer 发送 RequestVote RPC granted, term self._request_vote_rpc(self.node_id, self.current_term) if term self.current_term: # 发现更高的 term退为 Follower with self._lock: self.current_term term self.state NodeState.FOLLOWER self.voted_for None return if granted: votes 1 if votes majority: self._become_leader() return # 未过半退为 Follower with self._lock: if self.state NodeState.CANDIDATE: self.state NodeState.FOLLOWER def _request_vote_rpc(self, candidate_id: str, term: int) - Tuple[bool, int]: RequestVote RPC: 请求投票 Receiver 的实现模拟另一个节点处理投票请求 接收投票请求需要满足三个条件 1. candidate 的 term 当前节点的 term 2. 当前节点还没投过票或已投给这个 candidate 3. candidate 的日志至少和当前节点一样新 这里用简化模拟50% 15% term 优势的概率 if term self.current_term: return False, self.current_term # 模拟记录投票 with self._lock: if self.voted_for is None or self.voted_for candidate_id: # 检查日志是否足够新 last_log_term self.log[-1].term if self.log else 0 last_log_index len(self.log) self.voted_for candidate_id self.current_term term # 简化60% 概率同意更长的 term 有优势 accept_prob 0.6 min(0.3, (term - self.current_term) * 0.1) granted random.random() accept_prob return granted, self.current_term def _become_leader(self): 成为 Leader Leader 需要 1. 初始化每个 Follower 的 next_index 和 match_index 2. 立即发送一轮心跳AppendEntries 3. 启动心跳定时器 with self._lock: self.state NodeState.LEADER self.leader_id self.node_id # 初始化每个 Follower 的索引跟踪 last_log_index len(self.log) for peer_id in self.config.peers: self.next_index[peer_id] last_log_index 1 self.match_index[peer_id] 0 print(f[{self.node_id}] 成为 Leader (term{self.current_term})) # 立即发一轮心跳通知所有节点新的 Leader self._send_heartbeats() def _send_heartbeats(self): 向所有 Follower 发送心跳空的 AppendEntries 心跳有三个作用 1. 告知 Followers 自己还活着 2. 阻止 Followers 发起新选举 3. 附带 commit_index通知 Followers 哪些日志已提交 for peer_id in self.config.peers: success self._append_entries_rpc( peer_id, self.current_term, prev_log_indexlen(self.log), entries[] # 空 心跳 ) if not success: # Follower 拒绝了可能日志不匹配 # 实际实现递减 next_index 重试直到匹配 pass def _append_entries_rpc( self, follower_id: str, term: int, prev_log_index: int, entries: List[LogEntry] ) - bool: AppendEntries RPC: 心跳 日志复制 Leader 用这个 RPC 做两件事 1. 发送心跳entries 为空 2. 复制日志条目entries 非空 Receiver 的检查项 1. Leader 的 term 当前 term → 拒绝过时的 Leader 2. prev_log_index 处日志不匹配 → 拒绝需要追赶 3. 追加 entries截断冲突的后续日志 # 模拟80% 概率成功 return random.random() 0.8 def propose(self, command: str) - bool: 客户端向 Leader 提交命令 这是 Raft 的入口点。如果一个写操作想要生效 1. 必须提交到 Leader 2. Leader 复制到多数节点 3. Leader 提交并应用到状态机 如果当前节点不是 Leader返回 False。 调用方应该重定向到真正的 Leader。 if self.state ! NodeState.LEADER: print(f[{self.node_id}] 不是 Leader拒绝写入) return False with self._lock: entry LogEntry( termself.current_term, indexlen(self.log) 1, commandcommand ) self.log.append(entry) print(f[{self.node_id}] 提出命令: {command} (index{entry.index})) # 复制到 Followers replicated 0 total_nodes len(self.config.peers) 1 majority total_nodes // 2 1 for peer_id in self.config.peers: success self._append_entries_rpc( peer_id, self.current_term, prev_log_indexentry.index - 1, entries[entry] ) if success: replicated 1 replicated 1 # 自己的日志 if replicated majority: # 多数已复制 → 提交 self.commit_index entry.index self._apply_to_state_machine(entry) return True else: print(f[{self.node_id}] 未达到多数 (got {replicated}/{majority})) return False def _apply_to_state_machine(self, entry: LogEntry): 将已提交的日志条目应用到状态机 这是最终的一致状态。所有节点应用相同顺序的日志条目 因此状态机在所有节点上是确定性的。 if in entry.command: key, val entry.command.split(, 1) self._state_machine[key.strip()] val.strip() print(f[{self.node_id}] 状态机更新: {key} → {val}) def get_state(self) - dict: 获取节点当前状态用于监控 return { node_id: self.node_id, state: self.state.value, term: self.current_term, leader_id: self.leader_id, log_length: len(self.log), commit_index: self.commit_index, state_machine: dict(self._state_machine) } def simulate_election(): 模拟一次完整的 Leader 选举过程 # 创建 3 节点集群 configs [ RaftConfig(node_idworker-1, peers[worker-2, worker-3]), RaftConfig(node_idworker-2, peers[worker-1, worker-3]), RaftConfig(node_idworker-3, peers[worker-1, worker-2]), ] nodes [RaftNode(c) for c in configs] print(初始状态:) for n in nodes: s n.get_state() print(f {s[node_id]}: state{s[state]}, term{s[term]}) print(\n--- 节点 1 发起选举 ---) nodes[0].start_election() for n in nodes: s n.get_state() print(f {s[node_id]}: state{s[state]}, term{s[term]}, fleader{s[leader_id]}) print(\n--- Leader 写入数据 ---) leader next((n for n in nodes if n.state NodeState.LEADER), None) if leader: leader.propose(problem_42assigned_to_worker_07) leader.propose(problem_99assigned_to_worker_02) print(f\n最终状态机: {leader.get_state()[state_machine]}) if __name__ __main__: simulate_election()三、在评测集群中的应用在刷题系统的 Worker 集群中Raft 不是用来存数据数据在数据库里而是用来任务分配的一致性哪个 Worker 负责处理哪个题目。如果有多个 Leader 同时分配同一个题目的评测任务会导致重复评测或状态错乱。集群元数据管理当前集群有哪些 Worker 在线、每个 Worker 的负载状态。这些信息如果各节点各存一份脑裂时就失控。故障转移Leader Worker 挂了其他 Worker 通过选举自动接管任务分配。Raft 在评测集群的角色定位轻量级的分布式协调器类似 etcd 的角色不存储大数据量只保证关键元数据的一致性。四、边界Raft 不解决的问题不解决数据容量问题Raft 是复制所有状态到每个节点不适合大容量数据大容量用分片 Raft 的组合不解决读性能问题传统的 Raft 读操作也需要走 Leader 以确保线性一致性。如果读多写少应该用 leader lease 优化或让 Follower 读不解决跨数据中心延迟Raft 需要多数派响应才能提交跨机房的高延迟会直接拖慢写入五、总结Raft 是分布式系统的方向盘分布式系统中有两个永恒的难题谁是 Leader和哪些数据已提交。Raft 用一个统一的 term log index 机制同时回答了这两个问题。关键记住三点随机的选举超时不是 bug 而是 feature——正是它避免了同时选举的活锁Leader-based 复制简化了共识所有歧义由 Leader 裁决而非全节点协商心跳也是日志复制AppendEntries 既是心跳也是数据通道协议统一在 Judge Worker 集群中使用 Raft不是为了追赶技术潮流而是为了在节点故障时有一个确定的、可预见的故障恢复路径——这是同步协调手工重启 Leader无法提供的确定性。