上一讲我们实现了Raft的核心——Leader选举和日志复制。但在生产环境中还会遇到三个关键问题安全性如何保证不会选出已经落后的节点作为Leader成员变更如何在不停机的情况下增删节点日志压缩日志无限增长怎么办这一讲我们完成Raft的剩余部分。一、安全性保证1.1 Leader完整性Raft保证一旦某个日志条目在给定任期被提交它就会出现在所有未来任期的Leader中。# minikv/raft/safety.py import logging from typing import List, Optional from .types import * logger logging.getLogger(__name__) class SafetyChecker: Raft 安全性检查器 保证 1. Election Safety每个任期最多一个Leader 2. Leader Completeness已提交的日志一定存在于新Leader中 3. Log Matching两个节点的日志相同位置内容和任期相同 4. State Machine Safety状态机按相同顺序应用相同命令 staticmethod def check_log_matching(log1: List[LogEntry], log2: List[LogEntry]) - bool: 检查日志匹配特性 如果两个日志在相同索引位置的条目有相同的任期 那么它们在该索引之前的所有条目都相同 min_len min(len(log1), len(log2)) for i in range(min_len): if log1[i].term ! log2[i].term: return False return True staticmethod def is_log_up_to_date(my_last_term: int, my_last_index: int, candidate_last_term: int, candidate_last_index: int) - bool: 判断谁的日志更新 规则比较最后一个条目的任期任期更大的更新 任期相同则比较索引索引更大的更新 if my_last_term ! candidate_last_term: return my_last_term candidate_last_term return my_last_index candidate_last_index staticmethod def find_conflict_index(log: List[LogEntry], prev_log_index: int, prev_log_term: int) - tuple: 查找冲突位置 返回 (conflict_index, conflict_term) - 如果日志不够长返回缺失的索引 - 如果任期不匹配返回该任期的第一个索引 if prev_log_index len(log): return (len(log) 1, None) if prev_log_index 0: entry log[prev_log_index - 1] if entry.term ! prev_log_term: # 找到该任期的第一个条目 conflict_term entry.term idx prev_log_index - 1 while idx 0 and log[idx - 1].term conflict_term: idx - 1 return (idx 1, conflict_term) return (0, None)1.2 增强的选举逻辑# 在 raft_node.py 中增强 handle_request_vote def handle_request_vote(self, args: RequestVoteArgs, from_node: str) - RequestVoteResult: 增强的投票处理 增加安全性检查 1. 候选人日志必须至少和自己一样新 2. 每个任期只能投一票 with self.lock: result RequestVoteResult(termself.state.current_term, vote_grantedFalse) # 规则1任期检查 if args.term self.state.current_term: return result # 如果发现更高的任期转为Follower if args.term self.state.current_term: self.state.current_term args.term self.role NodeRole.FOLLOWER self.state.voted_for None # 规则2每个任期只能投一票 if self.state.voted_for is not None and self.state.voted_for ! args.candidate_id: return result # 规则3日志新旧检查Leader完整性保证 my_last_index len(self.state.log) my_last_term self.state.log[-1].term if self.state.log else 0 if not SafetyChecker.is_log_up_to_date( my_last_term, my_last_index, args.last_log_term, args.last_log_index ): return result # 投票 self.state.voted_for args.candidate_id result.vote_granted True self.election_deadline time.time() self._random_election_timeout() logger.info(f投票给 {args.candidate_id} (term{args.term}), f他的日志: term{args.last_log_term}, index{args.last_log_index}) return result二、成员变更2.1 联合共识Joint ConsensusRaft的成员变更采用两阶段协议保证变更过程中不会出现两个Leader阶段1Cold → Cnew - Leader 将 Cold,new 日志复制到所有节点 - 需要 Cold 和 Cnew 双多数派同意 阶段2Cnew - Leader 将 Cnew 日志复制到所有节点 - 只需 Cnew 多数派同意# minikv/raft/membership.py from dataclasses import dataclass, field from typing import List, Set, Optional import logging logger logging.getLogger(__name__) dataclass class ClusterConfig: 集群配置 nodes: Set[str] field(default_factoryset) joint_config: Optional[Set[str]] None # 联合共识时的旧配置 def is_joint(self) - bool: return self.joint_config is not None def get_quorum(self) - int: 获取当前配置的多数派大小 if self.is_joint(): # 需要两个配置各自的多数派 return max( len(self.nodes) // 2 1, len(self.joint_config) // 2 1 ) return len(self.nodes) // 2 1 def all_nodes(self) - Set[str]: 获取所有节点 if self.is_joint(): return self.nodes | self.joint_config return self.nodes class MembershipManager: 成员变更管理器 实现 Raft 的单节点变更一次只增删一个节点 和联合共识一次增删多个节点 def __init__(self, raft_node): self.raft raft_node self.config ClusterConfig(nodes{raft_node.node_id}) self.changing False def add_node(self, node_id: str) - bool: 添加节点 使用单节点变更一次只加一个节点 if node_id in self.config.nodes: logger.warning(f节点 {node_id} 已存在) return False if not self.raft.is_leader(): logger.warning(只有Leader才能变更成员) return False if self.changing: logger.warning(已有变更在进行中) return False self.changing True # 阶段1提交 Cnew 配置包含新节点 new_config self.config.nodes | {node_id} self._propose_config_change(new_config) return True def remove_node(self, node_id: str) - bool: 移除节点 if node_id not in self.config.nodes: logger.warning(f节点 {node_id} 不存在) return False if not self.raft.is_leader(): logger.warning(只有Leader才能变更成员) return False if self.changing: logger.warning(已有变更在进行中) return False self.changing True # 阶段1提交 Cnew 配置不含要移除的节点 new_config self.config.nodes - {node_id} self._propose_config_change(new_config) return True def _propose_config_change(self, new_nodes: Set[str]): 提议配置变更 # 如果是自己要被移除先转移Leader if self.raft.node_id not in new_nodes: # 选择一个新Leader target list(new_nodes)[0] self._transfer_leadership(target) return # 提交配置变更日志 config_entry LogEntry( indexlen(self.raft.state.log) 1, termself.raft.state.current_term, commandCONFIG_CHANGE, keycluster_config, valuelist(new_nodes) ) self.raft.state.log.append(config_entry) # 广播给所有节点 for peer in self.raft.peers: self.raft._send_append_entries(peer) logger.info(f配置变更提议: {new_nodes}) def apply_config_change(self, new_nodes: List[str]): 应用配置变更 old_nodes self.config.nodes.copy() self.config.nodes set(new_nodes) # 更新 peers 列表 self.raft.peers [n for n in new_nodes if n ! self.raft.node_id] self.changing False logger.info(f配置变更完成: {old_nodes} → {set(new_nodes)}) def _transfer_leadership(self, target: str): 转移领导权 # 发送TimeoutNow消息 # 目标节点收到后会立即发起选举 logger.info(f转移领导权给 {target}) # 简化直接停止自己的Leader角色 self.raft.role NodeRole.FOLLOWER三、日志压缩与快照3.1 快照机制当日志不断增长最终会耗尽内存。Raft通过快照来解决日志: [1] [2] [3] [4] [5] [6] [7] [8] ... ↓ 快照: snapshot(index5, term3, data{...}) 日志: [6] [7] [8] ...# minikv/raft/snapshot.py import pickle import os import threading import logging from typing import Any, Dict, Optional from dataclasses import dataclass logger logging.getLogger(__name__) dataclass class Snapshot: 快照 last_included_index: int 0 last_included_term: int 0 state_machine_data: Dict[str, Any] None def serialize(self) - bytes: return pickle.dumps(self) staticmethod def deserialize(data: bytes) - Snapshot: return pickle.loads(data) class SnapshotManager: 快照管理器 负责 1. 定期创建快照 2. 安装快照到落后节点 3. 截断日志 def __init__(self, raft_node, snapshot_dir: str ./snapshots, snapshot_threshold: int 10000): Args: raft_node: Raft节点实例 snapshot_dir: 快照存储目录 snapshot_threshold: 触发快照的日志条数阈值 self.raft raft_node self.snapshot_dir snapshot_dir self.snapshot_threshold snapshot_threshold self.last_snapshot: Optional[Snapshot] None os.makedirs(snapshot_dir, exist_okTrue) def maybe_take_snapshot(self): 检查是否需要创建快照 log_size len(self.raft.state.log) if log_size self.snapshot_threshold: self.take_snapshot() def take_snapshot(self): 创建快照 if not self.raft.state_machine: logger.warning(状态机未设置无法创建快照) return # 获取状态机数据 state_data self.raft.state_machine.get_snapshot_data() # 创建快照 last_index self.raft.volatile.last_applied last_term self.raft.state.log[last_index - 1].term if last_index 0 else 0 snapshot Snapshot( last_included_indexlast_index, last_included_termlast_term, state_machine_datastate_data ) # 保存快照 self._save_snapshot(snapshot) self.last_snapshot snapshot # 截断日志 self._truncate_log(last_index) logger.info(f创建快照: index{last_index}, term{last_term}, f日志从 {len(self.raft.state.log) last_index} 截断到 {len(self.raft.state.log)}) def install_snapshot(self, target: str): 向落后节点安装快照 当Follower落后太多时Leader直接发送快照而不是逐条同步日志 if not self.last_snapshot: return snapshot_data self.last_snapshot.serialize() # 发送InstallSnapshot RPC # (简化实现) logger.info(f向 {target} 安装快照: index{self.last_snapshot.last_included_index}) def apply_snapshot(self, snapshot: Snapshot): 应用快照到本节点 self.last_snapshot snapshot # 恢复状态机 if self.raft.state_machine: self.raft.state_machine.restore_from_snapshot(snapshot.state_machine_data) # 更新状态 self.raft.state.snapshot snapshot.serialize() self.raft.state.snapshot_index snapshot.last_included_index self.raft.state.snapshot_term snapshot.last_included_term # 截断日志 self._truncate_log(snapshot.last_included_index) logger.info(f应用快照: index{snapshot.last_included_index}) def _save_snapshot(self, snapshot: Snapshot): 保存快照到磁盘 filename fsnapshot_{snapshot.last_included_index}.bin filepath os.path.join(self.snapshot_dir, filename) with open(filepath, wb) as f: f.write(snapshot.serialize()) def _truncate_log(self, kept_index: int): 截断日志只保留kept_index之后的条目 if kept_index 0: self.raft.state.log self.raft.state.log[kept_index:]3.2 InstallSnapshot RPC# 添加到 raft_node.py dataclass class InstallSnapshotArgs: 安装快照参数 term: int leader_id: str last_included_index: int last_included_term: int data: bytes done: bool dataclass class InstallSnapshotResult: 安装快照结果 term: int def handle_install_snapshot(self, args: InstallSnapshotArgs) - InstallSnapshotResult: 处理安装快照请求 with self.lock: result InstallSnapshotResult(termself.state.current_term) if args.term self.state.current_term: return result if args.term self.state.current_term: self.state.current_term args.term self.role NodeRole.FOLLOWER self.state.voted_for None # 重置选举超时 self.election_deadline time.time() self._random_election_timeout() # 如果快照比当前日志更新 if args.last_included_index self.volatile.last_applied: # 反序列化快照 snapshot Snapshot.deserialize(args.data) # 应用快照 if self.snapshot_manager: self.snapshot_manager.apply_snapshot(snapshot) # 更新提交状态 self.volatile.commit_index max( self.volatile.commit_index, args.last_included_index ) self.volatile.last_applied max( self.volatile.last_applied, args.last_included_index ) return result四、持久化4.1 Raft状态持久化# minikv/raft/persistence.py import pickle import os import threading from typing import Optional from .types import RaftState class RaftPersister: Raft 状态持久化 保存和恢复 - currentTerm - votedFor - log - snapshot def __init__(self, node_id: str, data_dir: str ./raft_data): self.node_id node_id self.data_dir data_dir self.state_file os.path.join(data_dir, fraft_state_{node_id}.bin) self.lock threading.Lock() os.makedirs(data_dir, exist_okTrue) def save(self, state: RaftState): 保存Raft状态 with self.lock: with open(self.state_file, wb) as f: pickle.dump(state, f) def load(self) - Optional[RaftState]: 加载Raft状态 with self.lock: if not os.path.exists(self.state_file): return None with open(self.state_file, rb) as f: return pickle.load(f) def delete(self): 删除持久化状态 with self.lock: if os.path.exists(self.state_file): os.remove(self.state_file)五、集成到Raft节点# 增强 RaftNode 类 class EnhancedRaftNode(RaftNode): 增强的Raft节点包含完整功能 def __init__(self, node_id: str, peers: List[str], election_timeout: tuple (150, 300), heartbeat_interval: float 0.05, data_dir: str ./raft_data): super().__init__(node_id, peers, election_timeout, heartbeat_interval) # 持久化 self.persister RaftPersister(node_id, data_dir) # 快照管理 self.snapshot_manager SnapshotManager(self, snapshot_diros.path.join(data_dir, snapshots), snapshot_threshold1000 ) # 成员变更 self.membership MembershipManager(self) # 恢复状态 self._restore_state() def start(self): 启动节点增强版 # 恢复持久化状态 saved_state self.persister.load() if saved_state: self.state saved_state logger.info(f恢复Raft状态: term{self.state.current_term}, flog_size{len(self.state.log)}) super().start() # 启动快照检查 self._start_snapshot_checker() def _start_snapshot_checker(self): 启动快照检查 def check_loop(): while self.running: time.sleep(10) # 每10秒检查一次 self.snapshot_manager.maybe_take_snapshot() thread threading.Thread(targetcheck_loop, daemonTrue) thread.start() def persist_state(self): 持久化当前状态 self.persister.save(self.state) # 覆盖日志操作方法增加持久化 def _append_log(self, entry: LogEntry): 追加日志并持久化 self.state.log.append(entry) self.persist_state() def _truncate_log(self, index: int): 截断日志并持久化 self.state.log self.state.log[:index] self.persist_state()六、完整演示# examples/raft_full_demo.py import time import threading import logging import sys import os import tempfile logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s ) sys.path.insert(0, ..) from minikv.raft.raft_node import RaftNode from minikv.raft.safety import SafetyChecker from minikv.raft.snapshot import SnapshotManager, Snapshot from minikv.raft.membership import MembershipManager class SimpleStateMachine: 简单的状态机 def __init__(self): self.data {} def apply(self, command: str, key: str, valueNone): if command SET: self.data[key] value elif command DEL: self.data.pop(key, None) def get_snapshot_data(self) - dict: return self.data.copy() def restore_from_snapshot(self, data: dict): self.data data def demo_safety(): 演示安全性 print( * 60) print( Raft 安全性演示) print( * 60) # 创建模拟传输 transport SimulatedTransport() nodes {} for i in range(1, 4): node_id fnode-{i} peers [fnode-{j} for j in range(1, 4) if j ! i] raft RaftNode(node_id, peers, election_timeout(50, 100)) raft.state_machine SimpleStateMachine() transport.register(node_id, raft) raft.start() nodes[node_id] raft time.sleep(0.5) # 找到Leader并提交日志 leader [r for r in nodes.values() if r.is_leader()][0] leader_id leader.node_id print(f\nLeader: {leader_id}) leader.propose(SET, x, 100) leader.propose(SET, y, 200) time.sleep(0.3) print(\n 日志一致性检查:) logs [r.state.log for r in nodes.values()] consistent SafetyChecker.check_log_matching(logs[0], logs[1]) print(f node-1 vs node-2: {✅ 一致 if consistent else ❌ 不一致}) # 模拟Leader宕机 print(f\n {leader_id} 宕机) leader.stop() time.sleep(1) # 新Leader产生 new_leader [r for r in nodes.values() if r.running and r.is_leader()] if new_leader: nl new_leader[0] print(f\n 新Leader: {nl.node_id}) print(f 日志大小: {len(nl.state.log)}) print(f 已提交: {nl.volatile.commit_index}) # 验证已提交的日志仍然存在 safety_check SafetyChecker() for entry in nl.state.log: if entry.command SET: print(f 日志条目: {entry.command} {entry.key}{entry.value}) for raft in nodes.values(): raft.stop() def demo_snapshot(): 演示快照 print(\n * 60) print( 快照演示) print( * 60) with tempfile.TemporaryDirectory() as tmpdir: # 创建带快照功能的节点 raft RaftNode(node-1, [node-2, node-3]) raft.state_machine SimpleStateMachine() sm SnapshotManager(raft, snapshot_diros.path.join(tmpdir, snapshots), snapshot_threshold5 # 5条日志就触发快照 ) raft.snapshot_manager sm # 模拟大量日志 print(\n 模拟写入日志...) for i in range(20): entry LogEntry( indexi 1, term1, commandSET, keyfkey{i}, valuefvalue{i} ) raft.state.log.append(entry) raft.volatile.last_applied i 1 print(f 日志大小: {len(raft.state.log)}) # 创建快照 print(\n 创建快照...) sm.take_snapshot() print(f 快照后日志大小: {len(raft.state.log)}) print(f 快照索引: {sm.last_snapshot.last_included_index}) # 验证快照文件 snapshot_files os.listdir(sm.snapshot_dir) print(f 快照文件: {snapshot_files}) def demo_membership_change(): 演示成员变更 print(\n * 60) print( 成员变更演示) print( * 60) transport SimulatedTransport() nodes {} # 初始3节点 for i in range(1, 4): node_id fnode-{i} peers [fnode-{j} for j in range(1, 4) if j ! i] raft RaftNode(node_id, peers, election_timeout(50, 100)) transport.register(node_id, raft) raft.start() nodes[node_id] raft time.sleep(0.5) leader [r for r in nodes.values() if r.is_leader()] if leader: l leader[0] print(f\n当前Leader: {l.node_id}) print(f当前节点: {list(nodes.keys())}) # 模拟添加节点 print(\n➕ 添加 node-4...) membership MembershipManager(l) membership.add_node(node-4) time.sleep(0.3) print(f添加后配置: {membership.config.nodes}) for raft in nodes.values(): raft.stop() class SimulatedTransport: 模拟传输层 def __init__(self): self.nodes {} def register(self, node_id, raft): self.nodes[node_id] raft raft.send_message lambda t, mt, a: self._deliver(node_id, t, mt, a) def _deliver(self, sender, target, msg_type, args): if target in self.nodes: target_raft self.nodes[target] if msg_type RequestVote: result target_raft.handle_request_vote(args, sender) threading.Timer(0.01, self._deliver_vote_response, args[sender, target, result]).start() elif msg_type AppendEntries: result target_raft.handle_append_entries(args, sender) threading.Timer(0.01, self._deliver_append_response, args[sender, target, result]).start() def _deliver_vote_response(self, sender, target, result): if sender in self.nodes: self.nodes[sender].handle_vote_response(result, target) def _deliver_append_response(self, sender, target, result): if sender in self.nodes: self.nodes[sender].handle_append_entries_response(result, target) if __name__ __main__: demo_safety() demo_snapshot() demo_membership_change()七、测试# tests/test_raft_safety.py import unittest import time import threading from minikv.raft.raft_node import RaftNode from minikv.raft.safety import SafetyChecker from minikv.raft.snapshot import SnapshotManager, Snapshot class TestRaftSafety(unittest.TestCase): Raft安全性测试 def test_leader_completeness(self): 测试Leader完整性已提交的日志在新Leader中必须存在 transport SimulatedTransport() nodes {} for i in range(1, 4): node_id fnode-{i} peers [fnode-{j} for j in range(1, 4) if j ! i] raft RaftNode(node_id, peers, election_timeout(50, 100)) transport.register(node_id, raft) raft.start() nodes[node_id] raft time.sleep(0.5) # 找到Leader并提交日志 leader [r for r in nodes.values() if r.is_leader()][0] leader.propose(SET, committed_key, committed_value) time.sleep(0.3) # 记录已提交的日志 committed_entries list(leader.state.log) # 宕掉Leader leader.stop() time.sleep(1) # 新Leader产生 new_leader [r for r in nodes.values() if r.running and r.is_leader()] self.assertTrue(len(new_leader) 0) # 验证已提交的日志在新Leader中存在 nl new_leader[0] for committed in committed_entries: found any( e.index committed.index and e.term committed.term for e in nl.state.log ) self.assertTrue(found, f已提交的日志 {committed.index}/{committed.term} 丢失) for raft in nodes.values(): raft.stop() def test_log_matching(self): 测试日志匹配 log1 [ LogEntry(index1, term1, commandSET, keya), LogEntry(index2, term1, commandSET, keyb), LogEntry(index3, term2, commandSET, keyc), ] log2 [ LogEntry(index1, term1, commandSET, keya), LogEntry(index2, term1, commandSET, keyb), LogEntry(index3, term2, commandSET, keyc), ] self.assertTrue(SafetyChecker.check_log_matching(log1, log2)) # 不匹配的情况 log3 [ LogEntry(index1, term1, commandSET, keya), LogEntry(index2, term2, commandSET, keyx), # term不同 ] self.assertFalse(SafetyChecker.check_log_matching(log1, log3)) class TestSnapshot(unittest.TestCase): 快照测试 def test_snapshot_creation(self): 测试快照创建 import tempfile with tempfile.TemporaryDirectory() as tmpdir: raft RaftNode(test-node, []) sm SnapshotManager(raft, snapshot_dirtmpdir, snapshot_threshold5 ) # 填充日志 for i in range(10): raft.state.log.append(LogEntry( indexi 1, term1, commandSET, keyfk{i}, valuefv{i} )) raft.volatile.last_applied i 1 # 创建快照 sm.take_snapshot() # 验证快照文件 import os files os.listdir(tmpdir) self.assertTrue(len(files) 0) # 验证日志被截断 self.assertLess(len(raft.state.log), 10) class SimulatedTransport: def __init__(self): self.nodes {} def register(self, node_id, raft): self.nodes[node_id] raft raft.send_message lambda t, mt, a: self._deliver(node_id, t, mt, a) def _deliver(self, sender, target, msg_type, args): if target in self.nodes: target_raft self.nodes[target] if msg_type RequestVote: result target_raft.handle_request_vote(args, sender) threading.Timer(0.01, self._deliver_vote_response, args[sender, target, result]).start() elif msg_type AppendEntries: result target_raft.handle_append_entries(args, sender) threading.Timer(0.01, self._deliver_append_response, args[sender, target, result]).start() def _deliver_vote_response(self, sender, target, result): if sender in self.nodes: self.nodes[sender].handle_vote_response(result, target) def _deliver_append_response(self, sender, target, result): if sender in self.nodes: self.nodes[sender].handle_append_entries_response(result, target) if __name__ __main__: unittest.main()八、总结这一讲完成了Raft共识算法的全部核心功能功能实现安全性​Leader完整性、日志匹配、投票限制成员变更​单节点变更、联合共识、领导权转移快照​日志压缩、快照创建/安装、状态恢复持久化​Raft状态持久化、故障恢复至此MiniKV拥有了完整的分布式共识能力✅ 自动Leader选举✅ 可靠的日志复制✅ 安全的成员变更✅ 日志压缩防止无限增长✅ 故障恢复下一讲我们将把Raft集成到KV存储中实现一个基于Raft的分布式KV存储引擎——这才是MiniKV的真正核心。开发之余的小工具推荐​处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。