第5讲：基于Raft的分布式KV存储引擎

前四讲我们实现了节点通信、数据分片和Raft共识算法。现在是时候把它们整合起来了——这一讲我们将构建MiniKV的核心一个基于Raft的分布式KV存储引擎。这意味着客户端可以向任意节点写入数据Raft保证数据在所有节点上一致复制即使部分节点故障也不会丢失数据。一、整体架构┌─────────────────────────────────────┐ │ Client Request │ └────────────────┬────────────────────┘ │ ┌────────────────▼────────────────────┐ │ KV Service Layer │ │ (路由、转发、客户端API) │ └────────────────┬────────────────────┘ │ ┌────────────────▼────────────────────┐ │ Raft Consensus Layer │ │ (Leader选举、日志复制、成员变更) │ └────────────────┬────────────────────┘ │ ┌────────────────▼────────────────────┐ │ State Machine │ │ (实际的KV存储操作) │ └────────────────┬────────────────────┘ │ ┌────────────────▼────────────────────┐ │ Storage Engine │ │ (内存Map 可选持久化) │ └─────────────────────────────────────┘1.1 请求流程Client → 任意节点 ├─ 如果是Leader → 直接处理 │ ├─ 追加日志到Raft │ ├─ 复制到Follower │ ├─ 等待多数派确认 │ ├─ 应用到状态机 │ └─ 返回结果给Client │ └─ 如果是Follower → 转发给Leader └─ Leader处理完成后返回结果二、KV状态机2.1 状态机接口# minikv/kv/state_machine.py from typing import Any, Dict, List, Optional, Tuple from dataclasses import dataclass, field import threading import json dataclass class Command: 状态机命令 operation: str # GET, SET, DELETE, CAS key: str value: Any None expected_value: Any None # 用于CAS操作 dataclass class CommandResult: 命令执行结果 success: bool True value: Any None error: str redirect: str # 如果不是Leader重定向地址 class KVStateMachine: KV状态机 这是Raft复制的状态机所有写操作必须通过Raft日志 读操作可以直接从状态机读取线性一致性读需要特殊处理 def __init__(self): self.data: Dict[str, Any] {} self.lock threading.RLock() self.version: Dict[str, int] {} # 每个key的版本号 # 监听器 self.watch_callbacks: List[callable] [] def apply(self, command: Command) - CommandResult: 应用命令到状态机 这个方法只能由Raft的apply循环调用 with self.lock: if command.operation SET: return self._apply_set(command) elif command.operation DELETE: return self._apply_delete(command) elif command.operation CAS: return self._apply_cas(command) else: return CommandResult(successFalse, errorf未知操作: {command.operation}) def read(self, key: str) - Optional[Any]: 读取数据直接读不经过Raft 注意要实现线性一致性需要在读取时也经过Raft with self.lock: return self.data.get(key) def read_all(self) - Dict[str, Any]: 读取所有数据 with self.lock: return self.data.copy() def _apply_set(self, command: Command) - CommandResult: 应用SET命令 old_value self.data.get(command.key) self.data[command.key] command.value self.version[command.key] self.version.get(command.key, 0) 1 self._notify_watch(command.key, SET, old_value, command.value) return CommandResult(successTrue, valuecommand.value) def _apply_delete(self, command: Command) - CommandResult: 应用DELETE命令 old_value self.data.pop(command.key, None) if old_value is not None: self.version[command.key] self.version.get(command.key, 0) 1 self._notify_watch(command.key, DELETE, old_value, None) return CommandResult(successTrue) return CommandResult(successFalse, errorKey not found) def _apply_cas(self, command: Command) - CommandResult: 应用CASCompare-And-Swap命令 current self.data.get(command.key) if current command.expected_value: self.data[command.key] command.value self.version[command.key] self.version.get(command.key, 0) 1 self._notify_watch(command.key, CAS, current, command.value) return CommandResult(successTrue, valuecommand.value) return CommandResult( successFalse, errorfCAS失败: 期望 {command.expected_value}, 实际 {current}, valuecurrent ) def get_snapshot_data(self) - Dict[str, Any]: 获取快照数据 with self.lock: return { data: self.data.copy(), version: self.version.copy() } def restore_from_snapshot(self, data: Dict[str, Any]): 从快照恢复 with self.lock: self.data data.get(data, {}) self.version data.get(version, {}) def watch(self, callback: callable): 添加watch回调 self.watch_callbacks.append(callback) def _notify_watch(self, key: str, operation: str, old_value: Any, new_value: Any): 通知watch监听器 for callback in self.watch_callbacks: try: callback(key, operation, old_value, new_value) except Exception as e: print(fWatch callback error: {e})三、分布式KV服务3.1 KV服务实现# minikv/kv/kv_service.py import threading import time import logging from typing import Dict, List, Optional, Any from ..raft.raft_node import EnhancedRaftNode from ..raft.types import LogEntry from .state_machine import KVStateMachine, Command, CommandResult logger logging.getLogger(__name__) class KVService: 分布式KV服务 提供 - 完整的KV APIGET/SET/DELETE/CAS - 自动路由到Leader - 线性一致性读 - Watch机制 def __init__(self, node_id: str, peers: List[str], data_dir: str ./kv_data): self.node_id node_id self.peers peers # 状态机 self.state_machine KVStateMachine() # Raft节点 self.raft EnhancedRaftNode( node_idnode_id, peerspeers, data_dirdata_dir ) self.raft.state_machine self.state_machine # 传输层由外部注入 self.transport None # 缓存Leader信息 self.leader_id: Optional[str] None self.leader_addr: Optional[tuple] None # 等待完成的请求 self.pending_requests: Dict[str, threading.Event] {} self.pending_results: Dict[str, CommandResult] {} # 启动时读取 self.ready threading.Event() def start(self): 启动服务 # 启动Raft self.raft.start() # 启动后台任务 self._start_background_tasks() self.ready.set() logger.info(fKVService[{self.node_id}] started) def stop(self): 停止服务 self.raft.stop() logger.info(fKVService[{self.node_id}] stopped) # 客户端API def get(self, key: str, linearizable: bool True) - CommandResult: 读取key的值 Args: key: 要读取的key linearizable: 是否使用线性一致性读 True: 经过Raft保证读到最新值 False: 直接从状态机读可能读到旧值 if not self.ready.is_set(): return CommandResult(successFalse, error服务未就绪) # 如果不是Leader转发 if not self.raft.is_leader(): if self.leader_id and self.leader_id ! self.node_id: return CommandResult( successFalse, errorNot leader, redirectself.leader_id ) return CommandResult(successFalse, errorNot leader) if linearizable: # 线性一致性读通过Raft进行一次空提交确保读到最新数据 # 这里简化为直接读取 等待一次心跳 self._ensure_read_index() value self.state_machine.read(key) return CommandResult(successTrue, valuevalue) def set(self, key: str, value: Any) - CommandResult: 设置key的值 这是一个写操作必须经过Raft if not self.ready.is_set(): return CommandResult(successFalse, error服务未就绪) # 如果不是Leader转发 if not self.raft.is_leader(): if self.leader_id: return CommandResult( successFalse, errorNot leader, redirectself.leader_id ) return CommandResult(successFalse, errorNot leader) # 创建命令并通过Raft提交 command Command(operationSET, keykey, valuevalue) return self._propose_command(command) def delete(self, key: str) - CommandResult: 删除key if not self.raft.is_leader(): return CommandResult(successFalse, errorNot leader, redirectself.leader_id) command Command(operationDELETE, keykey) return self._propose_command(command) def cas(self, key: str, expected: Any, new_value: Any) - CommandResult: Compare-And-Swap 原子操作只有当key的当前值等于expected时才更新 if not self.raft.is_leader(): return CommandResult(successFalse, errorNot leader, redirectself.leader_id) command Command( operationCAS, keykey, valuenew_value, expected_valueexpected ) return self._propose_command(command) # 内部方法 def _propose_command(self, command: Command) - CommandResult: 通过Raft提交命令 1. 追加到Raft日志 2. 等待日志被提交 3. 状态机应用后返回结果 # 生成请求ID request_id f{self.node_id}_{time.time_ns()} # 注册等待 event threading.Event() self.pending_requests[request_id] event # 通过Raft提议 success self.raft.propose( command.operation, command.key, command.value ) if not success: del self.pending_requests[request_id] return CommandResult(successFalse, errorPropose failed) # 等待结果超时5秒 if event.wait(timeout5.0): result self.pending_results.pop(request_id, CommandResult(successFalse, errorResult lost)) del self.pending_requests[request_id] return result else: del self.pending_requests[request_id] return CommandResult(successFalse, errorTimeout) def _on_command_applied(self, command: Command, result: CommandResult): 命令被应用到状态机后的回调 由Raft的apply循环调用 # 通知等待的请求 for request_id, event in list(self.pending_requests.items()): # 简化通知所有等待的请求 # 实际应该根据请求ID精确匹配 self.pending_results[request_id] result event.set() def _ensure_read_index(self): 确保读取索引线性一致性读辅助 # 通过一次空的心跳确认 # 简化实现等待一小段时间 time.sleep(0.001) def _start_background_tasks(self): 启动后台任务 def monitor_leader(): 监控Leader变化 while self.raft.running: if self.raft.is_leader(): self.leader_id self.node_id else: # 从Raft状态获取Leader信息 # (简化通过心跳检测) pass time.sleep(0.1) thread threading.Thread(targetmonitor_leader, daemonTrue) thread.start() def handle_redirect(self, result: CommandResult) - Optional[str]: 处理重定向 如果当前节点不是Leader返回Leader地址 if result.redirect: return result.redirect return None四、客户端实现4.1 分布式KV客户端# minikv/kv/client.py import random import time import logging from typing import List, Optional, Any, Dict from .state_machine import CommandResult logger logging.getLogger(__name__) class KVClient: 分布式KV客户端 特性 - 自动发现集群节点 - 自动重定向到Leader - 请求重试 - 连接池 def __init__(self, seeds: List[str], max_retries: int 3, timeout: float 5.0): Args: seeds: 种子节点列表 [node1:port, node2:port, ...] max_retries: 最大重试次数 timeout: 请求超时时间秒 self.seeds seeds self.max_retries max_retries self.timeout timeout # 节点连接 self.connections: Dict[str, any] {} # 当前已知的Leader self.leader: Optional[str] None # 服务引用简化直接持有KVService self.services: Dict[str, KVService] {} def connect(self, services: Dict[str, KVService]): 连接到集群 self.services services logger.info(fConnected to cluster with {len(services)} nodes) def get(self, key: str) - Optional[Any]: 读取key for attempt in range(self.max_retries): try: # 选择目标节点 target self._select_target() service self.services.get(target) if not service: continue result service.get(key) if result.success: return result.value # 如果需要重定向 if result.redirect: self.leader result.redirect continue if result.error: logger.warning(fGET {key} failed: {result.error}) return None except Exception as e: logger.error(fGET {key} error (attempt {attempt 1}): {e}) time.sleep(0.1 * (attempt 1)) return None def set(self, key: str, value: Any) - bool: 设置key for attempt in range(self.max_retries): try: target self._select_target() service self.services.get(target) if not service: continue result service.set(key, value) if result.success: return True if result.redirect: self.leader result.redirect continue except Exception as e: logger.error(fSET {key} error (attempt {attempt 1}): {e}) time.sleep(0.1 * (attempt 1)) return False def delete(self, key: str) - bool: 删除key for attempt in range(self.max_retries): try: target self._select_target() service self.services.get(target) if not service: continue result service.delete(key) if result.success: return True if result.redirect: self.leader result.redirect continue except Exception as e: logger.error(fDELETE {key} error (attempt {attempt 1}): {e}) time.sleep(0.1 * (attempt 1)) return False def cas(self, key: str, expected: Any, new_value: Any) - bool: Compare-And-Swap for attempt in range(self.max_retries): try: target self._select_target() service self.services.get(target) if not service: continue result service.cas(key, expected, new_value) if result.success: return True if result.redirect: self.leader result.redirect continue except Exception as e: logger.error(fCAS {key} error (attempt {attempt 1}): {e}) time.sleep(0.1 * (attempt 1)) return False def _select_target(self) - str: 选择目标节点 # 优先使用已知的Leader if self.leader and self.leader in self.services: return self.leader # 随机选择一个节点 return random.choice(list(self.services.keys()))五、集成与启动5.1 集群启动器# minikv/kv/cluster.py import threading import time import logging from typing import Dict, List from .kv_service import KVService from .client import KVClient logger logging.getLogger(__name__) class MiniKVCluster: MiniKV 集群管理器 负责启动和管理整个集群 def __init__(self, node_count: int 3, base_port: int 9000, data_dir: str ./kv_data): self.node_count node_count self.base_port base_port self.data_dir data_dir self.nodes: Dict[str, KVService] {} self.client: KVClient None def start(self): 启动集群 logger.info(fStarting MiniKV cluster with {self.node_count} nodes) # 生成节点ID node_ids [fnode-{i 1} for i in range(self.node_count)] # 启动每个节点 for i, node_id in enumerate(node_ids): peers [n for n in node_ids if n ! node_id] service KVService( node_idnode_id, peerspeers, data_dirf{self.data_dir}/{node_id} ) self.nodes[node_id] service service.start() logger.info(f Started {node_id}) time.sleep(0.2) # 错开启动时间避免选举冲突 # 等待Leader选举 self._wait_for_leader() # 创建客户端 self.client KVClient(seedsnode_ids) self.client.connect(self.nodes) logger.info(Cluster ready!) return self.client def stop(self): 停止集群 logger.info(Stopping cluster...) for node_id, service in self.nodes.items(): service.stop() logger.info(Cluster stopped) def get_leader(self) - str: 获取当前Leader for node_id, service in self.nodes.items(): if service.raft.is_leader(): return node_id return None def kill_node(self, node_id: str): 杀掉一个节点模拟故障 if node_id in self.nodes: logger.warning(fKilling node: {node_id}) self.nodes[node_id].stop() def add_node(self, node_id: str): 添加新节点 if node_id in self.nodes: return peers [n for n in self.nodes.keys()] service KVService( node_idnode_id, peerspeers, data_dirf{self.data_dir}/{node_id} ) self.nodes[node_id] service service.start() logger.info(fAdded node: {node_id}) def _wait_for_leader(self, timeout: float 10.0): 等待Leader选举完成 start time.time() while time.time() - start timeout: leader self.get_leader() if leader: logger.info(fLeader elected: {leader}) return leader time.sleep(0.1) raise TimeoutError(Leader election timeout) def get_status(self) - dict: 获取集群状态 status { node_count: len(self.nodes), leader: self.get_leader(), nodes: {} } for node_id, service in self.nodes.items(): raft_state service.raft.get_state() status[nodes][node_id] { role: raft_state[role], term: raft_state[current_term], log_size: raft_state[log_size], commit_index: raft_state[commit_index], data_size: len(service.state_machine.data) } return status六、完整演示# examples/kv_cluster_demo.py import time 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.kv.cluster import MiniKVCluster def demo_basic_operations(): 演示基本KV操作 print( * 60) print( MiniKV 分布式KV存储演示) print( * 60) with tempfile.TemporaryDirectory() as tmpdir: # 启动3节点集群 print(\n 启动3节点集群...) cluster MiniKVCluster( node_count3, base_port9100, data_diros.path.join(tmpdir, kv_data) ) client cluster.start() # 基本操作 print(\n 基本KV操作:) # SET print(\n SET operations:) assert client.set(name, Alice) print( name Alice ✅) assert client.set(age, 30) print( age 30 ✅) assert client.set(city, Beijing) print( city Beijing ✅) # GET print(\n GET operations:) name client.get(name) print(f name {name} ✅) assert name Alice age client.get(age) print(f age {age} ✅) assert age 30 # 不存在的key missing client.get(nonexistent) print(f nonexistent {missing} ✅) assert missing is None # DELETE print(\n DELETE operations:) assert client.delete(city) print( city deleted ✅) deleted client.get(city) print(f city after delete {deleted} ✅) assert deleted is None # CAS print(\n CAS operations:) assert client.cas(age, 30, 31) print( CAS age from 30 to 31 ✅) new_age client.get(age) print(f age after CAS {new_age} ✅) assert new_age 31 # CAS失败 result client.cas(age, 30, 32) # 当前值是31 print(f CAS with wrong expected value {result} ✅) assert result False # 集群状态 print(\n 集群状态:) status cluster.get_status() print(f Leader: {status[leader]}) for node_id, info in status[nodes].items(): print(f {node_id}: {info[role]}, fterm{info[term]}, fdata{info[data_size]} keys) cluster.stop() def demo_fault_tolerance(): 演示容错性 print(\n * 60) print(️ 容错性演示) print( * 60) with tempfile.TemporaryDirectory() as tmpdir: cluster MiniKVCluster( node_count3, base_port9200, data_diros.path.join(tmpdir, kv_data) ) client cluster.start() # 写入一些数据 print(\n 写入初始数据...) for i in range(10): client.set(fkey{i}, fvalue{i}) print( 10 keys written ✅) # 记录Leader leader cluster.get_leader() print(f\n当前Leader: {leader}) # 杀掉Leader print(f\n 杀掉Leader节点: {leader}) cluster.kill_node(leader) time.sleep(2) # 等待重新选举 # 检查新Leader new_leader cluster.get_leader() print(f新Leader: {new_leader}) assert new_leader ! leader # 验证数据仍然可用 print(\n✅ 验证数据完整性:) all_ok True for i in range(10): value client.get(fkey{i}) if value ! fvalue{i}: print(f key{i}: 期望 value{i}, 得到 {value} ❌) all_ok False else: print(f key{i} {value} ✅) assert all_ok, 数据完整性检查失败! # 继续写入 print(\n 继续写入新数据...) client.set(new_key, new_value) value client.get(new_key) print(f new_key {value} ✅) cluster.stop() def demo_linearizability(): 演示线性一致性 print(\n * 60) print( 线性一致性演示) print( * 60) with tempfile.TemporaryDirectory() as tmpdir: cluster MiniKVCluster( node_count3, base_port9300, data_diros.path.join(tmpdir, kv_data) ) client cluster.start() # 模拟并发读写 import threading import random results [] lock threading.Lock() def writer(): for i in range(20): client.set(fcounter, i) time.sleep(0.01) def reader(): for i in range(50): value client.get(counter) with lock: results.append(value) time.sleep(0.005) threads [] for _ in range(3): t threading.Thread(targetreader) threads.append(t) t.start() w threading.Thread(targetwriter) w.start() for t in threads: t.join() w.join() print(f\n 读取到的值共 {len(results)} 次:) print(f 最小值: {min(results)}) print(f 最大值: {max(results)}) print(f 单调递增: {results sorted(results)}) cluster.stop() if __name__ __main__: demo_basic_operations() demo_fault_tolerance() demo_linearizability()七、测试# tests/test_kv_service.py import unittest import time import tempfile import os from minikv.kv.cluster import MiniKVCluster from minikv.kv.state_machine import KVStateMachine, Command class TestKVStateMachine(unittest.TestCase): KV状态机测试 def setUp(self): self.sm KVStateMachine() def test_set_and_get(self): 测试SET和GET cmd Command(operationSET, keyname, valueAlice) result self.sm.apply(cmd) self.assertTrue(result.success) self.assertEqual(self.sm.read(name), Alice) def test_delete(self): 测试DELETE self.sm.apply(Command(operationSET, keyx, value1)) result self.sm.apply(Command(operationDELETE, keyx)) self.assertTrue(result.success) self.assertIsNone(self.sm.read(x)) def test_cas_success(self): 测试CAS成功 self.sm.apply(Command(operationSET, keycounter, value10)) result self.sm.apply(Command( operationCAS, keycounter, expected_value10, new_value11 )) self.assertTrue(result.success) self.assertEqual(self.sm.read(counter), 11) def test_cas_failure(self): 测试CAS失败 self.sm.apply(Command(operationSET, keycounter, value10)) result self.sm.apply(Command( operationCAS, keycounter, expected_value99, new_value11 )) self.assertFalse(result.success) self.assertEqual(self.sm.read(counter), 10) def test_snapshot(self): 测试快照 self.sm.apply(Command(operationSET, keya, value1)) self.sm.apply(Command(operationSET, keyb, value2)) snapshot self.sm.get_snapshot_data() new_sm KVStateMachine() new_sm.restore_from_snapshot(snapshot) self.assertEqual(new_sm.read(a), 1) self.assertEqual(new_sm.read(b), 2) class TestKVCluster(unittest.TestCase): KV集群测试 def setUp(self): self.tmpdir tempfile.mkdtemp() self.cluster MiniKVCluster( node_count3, base_port9400, data_diros.path.join(self.tmpdir, kv_data) ) self.client self.cluster.start() def tearDown(self): self.cluster.stop() def test_basic_operations(self): 测试基本操作 self.assertTrue(self.client.set(key1, value1)) self.assertEqual(self.client.get(key1), value1) self.assertTrue(self.client.delete(key1)) self.assertIsNone(self.client.get(key1)) def test_fault_tolerance(self): 测试容错 # 写入数据 for i in range(5): self.client.set(fk{i}, fv{i}) # 杀掉Leader leader self.cluster.get_leader() self.cluster.kill_node(leader) time.sleep(2) # 验证数据 for i in range(5): self.assertEqual(self.client.get(fk{i}), fv{i}) def test_consistency(self): 测试一致性 # 所有节点最终应该看到相同的数据 self.client.set(shared, data) time.sleep(0.5) for node_id, service in self.cluster.nodes.items(): value service.state_machine.read(shared) self.assertEqual(value, data, fNode {node_id} inconsistent) if __name__ __main__: unittest.main()八、总结这一讲我们构建了MiniKV的核心——基于Raft的分布式KV存储引擎组件功能KV状态机​状态转换、快照、Watch机制KV服务​读写API、Raft集成、Leader转发KV客户端​自动发现、重试、重定向集群管理​启动、停止、故障注入、状态查询关键成果✅ 完整的分布式KV APIGET/SET/DELETE/CAS✅ 自动Leader选举和请求转发✅ 节点故障时数据不丢失✅ 线性一致性读✅ 快照备份与恢复下一讲我们将实现分布式事务——让MiniKV支持跨多个key的原子操作。开发之余的小工具推荐​处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。