到目前为止MiniKV支持单个key的原子操作SET/GET/DELETE/CAS但现实业务往往需要同时操作多个key——比如转账需要同时扣减账户A和增加账户B。这一讲我们为MiniKV实现分布式事务支持跨多个key的ACID操作。一、设计思路1.1 为什么需要分布式事务转账场景 account_A - 100 account_B 100 如果不使用事务可能出现 1. account_A 扣减成功account_B 增加失败 → 钱丢了 2. 中途读取到中间状态 → 数据不一致1.2 事务隔离级别MiniKV实现可串行化快照隔离SSI隔离级别脏读不可重复读幻读实现方式Read Committed❌✅✅简单Repeatable Read❌❌✅MVCCSerializable❌❌❌SSI/OCC1.3 事务模型MiniKV采用乐观并发控制OCCBEGIN TX READ phase: 读取所有需要的key记录版本号 WRITE phase: 在本地缓冲区写入修改 COMMIT: 1. 检查所有读取的key版本号是否变化冲突检测 2. 如果没有冲突通过Raft提交事务日志 3. 如果有冲突事务回滚二、事务核心数据结构# minikv/txn/types.py from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Set import time import uuid from enum import Enum, auto class TxnStatus(Enum): PENDING auto() # 事务进行中 COMMITTED auto() # 已提交 ABORTED auto() # 已中止 STALE auto() # 过期长时间未提交 dataclass class ReadRecord: 读取记录 key: str value: Any version: int # 读取时的版本号 timestamp: float # 读取时间 dataclass class WriteRecord: 写入记录 key: str value: Any is_delete: bool False dataclass class Transaction: 事务 txn_id: str # 事务ID status: TxnStatus TxnStatus.PENDING reads: Dict[str, ReadRecord] field(default_factorydict) # 读取集 writes: Dict[str, WriteRecord] field(default_factorydict) # 写入集 start_ts: float 0.0 # 开始时间戳 commit_ts: float 0.0 # 提交时间戳 def __post_init__(self): if not self.txn_id: self.txn_id str(uuid.uuid4()) if not self.start_ts: self.start_ts time.time() dataclass class TxnCommand: 事务命令通过Raft复制 txn_id: str writes: Dict[str, WriteRecord] reads: Dict[str, int] # key - 读取时的版本号 commit_ts: float dataclass class TxnResult: 事务执行结果 success: bool True txn_id: str values: Dict[str, Any] field(default_factorydict) error: str retry: bool False # 是否应该重试三、事务管理器3.1 本地事务管理# minikv/txn/manager.py import threading import time import logging from typing import Dict, List, Optional, Any, Set from .types import * from ..kv.state_machine import KVStateMachine logger logging.getLogger(__name__) class TxnManager: 事务管理器 负责 1. 事务生命周期管理 2. 冲突检测 3. 事务提交/回滚 def __init__(self, state_machine: KVStateMachine, txn_timeout: float 30.0): self.state_machine state_machine self.txn_timeout txn_timeout # 活跃事务 self.active_txns: Dict[str, Transaction] {} # 已提交事务的时间戳用于GC self.committed_timestamps: List[float] [] # 锁 self.lock threading.RLock() # 清理线程 self._start_cleanup_thread() def begin(self) - Transaction: 开启一个新事务 txn Transaction() with self.lock: self.active_txns[txn.txn_id] txn logger.debug(fBegin transaction: {txn.txn_id}) return txn def read(self, txn: Transaction, key: str) - Optional[Any]: 在事务中读取key 先检查写入缓冲区再读取状态机 # 检查写入缓冲区 if key in txn.writes: write txn.writes[key] if write.is_delete: return None return write.value # 从状态机读取 value self.state_machine.read(key) version self.state_machine.version.get(key, 0) # 记录读取 txn.reads[key] ReadRecord( keykey, valuevalue, versionversion, timestamptime.time() ) return value def write(self, txn: Transaction, key: str, value: Any): 在事务中写入key写入缓冲区 txn.writes[key] WriteRecord(keykey, valuevalue) def delete(self, txn: Transaction, key: str): 在事务中删除key txn.writes[key] WriteRecord(keykey, valueNone, is_deleteTrue) def commit(self, txn: Transaction) - TxnResult: 提交事务 使用乐观并发控制 1. 检查所有读取的key版本号是否变化 2. 如果没有冲突生成TxnCommand并通过Raft提交 3. 如果有冲突回滚事务 with self.lock: if txn.status ! TxnStatus.PENDING: return TxnResult( successFalse, txn_idtxn.txn_id, errorfInvalid status: {txn.status} ) # 冲突检测 conflict self._detect_conflicts(txn) if conflict: txn.status TxnStatus.ABORTED self.active_txns.pop(txn.txn_id, None) return TxnResult( successFalse, txn_idtxn.txn_id, errorfConflict detected on key: {conflict}, retryTrue ) # 标记事务为提交中 txn.status TxnStatus.COMMITTED txn.commit_ts time.time() # 生成提交命令 reads_version { key: record.version for key, record in txn.reads.items() } command TxnCommand( txn_idtxn.txn_id, writestxn.writes, readsreads_version, commit_tstxn.commit_ts ) # 从活跃事务中移除 self.active_txns.pop(txn.txn_id, None) self.committed_timestamps.append(txn.commit_ts) return TxnResult( successTrue, txn_idtxn.txn_id, values{ key: write.value for key, write in txn.writes.items() if not write.is_delete } ) def rollback(self, txn: Transaction): 回滚事务 with self.lock: txn.status TxnStatus.ABORTED self.active_txns.pop(txn.txn_id, None) logger.debug(fRollback transaction: {txn.txn_id}) def apply_txn_command(self, command: TxnCommand): 应用事务命令到状态机 由Raft的apply循环调用 with self.lock: # 再次检查冲突防止并发提交 for key, version in command.reads.items(): current_version self.state_machine.version.get(key, 0) if current_version ! version: logger.warning(fTransaction {command.txn_id} conflict on {key}) return False # 应用所有写入 for key, write in command.writes.items(): if write.is_delete: # 直接操作状态机内部数据 self.state_machine.data.pop(key, None) else: self.state_machine.data[key] write.value self.state_machine.version[key] \ self.state_machine.version.get(key, 0) 1 logger.info(fTransaction {command.txn_id} applied) return True def _detect_conflicts(self, txn: Transaction) - Optional[str]: 检测读写冲突 检查事务读取的key在当前是否有其他事务正在写入 for key, record in txn.reads.items(): current_version self.state_machine.version.get(key, 0) if current_version ! record.version: return key # 检查写写冲突 for key in txn.writes: for other_txn in self.active_txns.values(): if other_txn.txn_id txn.txn_id: continue if key in other_txn.writes: return f{key} (concurrent write) return None def _start_cleanup_thread(self): 启动清理线程清理过期事务和旧时间戳 def cleanup(): while True: time.sleep(60) # 每分钟清理一次 with self.lock: now time.time() # 清理过期事务 stale_txns [ tid for tid, txn in self.active_txns.items() if now - txn.start_ts self.txn_timeout ] for tid in stale_txns: self.active_txns[tid].status TxnStatus.STALE self.active_txns.pop(tid) # 清理旧时间戳保留最近100个 if len(self.committed_timestamps) 100: self.committed_timestamps \ self.committed_timestamps[-100:] thread threading.Thread(targetcleanup, daemonTrue) thread.start() def get_active_txns(self) - List[dict]: 获取活跃事务列表 with self.lock: return [ { txn_id: txn.txn_id, status: txn.status.name, reads: list(txn.reads.keys()), writes: list(txn.writes.keys()), duration: time.time() - txn.start_ts } for txn in self.active_txns.values() ]四、事务服务层4.1 集成到KV服务# minikv/txn/txn_service.py import threading import time import logging from typing import Dict, List, Optional, Any from .types import * from .manager import TxnManager from ..kv.state_machine import KVStateMachine, Command, CommandResult logger logging.getLogger(__name__) class TxnService: 事务服务 提供事务API集成Raft和状态机 def __init__(self, state_machine: KVStateMachine, raft_node): self.state_machine state_machine self.raft raft_node self.txn_manager TxnManager(state_machine) # 挂起的事务等待Raft提交 self.pending_txns: Dict[str, threading.Event] {} self.pending_results: Dict[str, TxnResult] {} # 设置Raft的状态机回调 self._setup_raft_callback() def begin(self) - Transaction: 开启事务 return self.txn_manager.begin() def read(self, txn: Transaction, key: str) - Any: 事务内读取 return self.txn_manager.read(txn, key) def write(self, txn: Transaction, key: str, value: Any): 事务内写入 self.txn_manager.write(txn, key, value) def delete(self, txn: Transaction, key: str): 事务内删除 self.txn_manager.delete(txn, key) def commit(self, txn: Transaction) - TxnResult: 提交事务 流程 1. 冲突检测本地 2. 生成TxnCommand 3. 通过Raft复制 4. 应用到状态机 # 本地冲突检测 result self.txn_manager.commit(txn) if not result.success: return result # 生成Raft命令 reads_version { key: record.version for key, record in txn.reads.items() } command TxnCommand( txn_idtxn.txn_id, writestxn.writes, readsreads_version, commit_tstxn.commit_ts ) # 通过Raft提交 return self._propose_txn_command(command) def rollback(self, txn: Transaction): 回滚事务 self.txn_manager.rollback(txn) def _propose_txn_command(self, command: TxnCommand) - TxnResult: 通过Raft提交事务命令 # 注册等待 event threading.Event() self.pending_txns[command.txn_id] event # 通过Raft提议 success self.raft.propose( TX_COMMIT, command.txn_id, command ) if not success: del self.pending_txns[command.txn_id] return TxnResult( successFalse, txn_idcommand.txn_id, errorPropose failed ) # 等待结果 if event.wait(timeout5.0): result self.pending_results.pop(command.txn_id, TxnResult(successFalse, errorResult lost)) del self.pending_txns[command.txn_id] return result else: del self.pending_txns[command.txn_id] return TxnResult( successFalse, txn_idcommand.txn_id, errorTimeout, retryTrue ) def _setup_raft_callback(self): 设置Raft回调 original_apply self.state_machine.apply def txn_aware_apply(command: Command): if command.operation TX_COMMIT: txn_cmd command.value success self.txn_manager.apply_txn_command(txn_cmd) # 通知等待的请求 if txn_cmd.txn_id in self.pending_txns: result TxnResult( successsuccess, txn_idtxn_cmd.txn_id ) self.pending_results[txn_cmd.txn_id] result self.pending_txns[txn_cmd.txn_id].set() return CommandResult(successsuccess) else: return original_apply(command) self.state_machine.apply txn_aware_apply五、事务客户端5.1 高级事务API# minikv/txn/client.py import logging from typing import Any, Dict, List, Optional, Callable from .types import Transaction, TxnResult from .txn_service import TxnService logger logging.getLogger(__name__) class TxnClient: 事务客户端 提供高级事务API包括 - 自动重试 - 上下文管理器 - 回调模式 def __init__(self, txn_service: TxnService, max_retries: int 3): self.txn_service txn_service self.max_retries max_retries def run(self, txn_func: Callable[[Transaction], Any]) - Any: 运行一个事务函数 自动处理重试和冲突 Usage: def transfer(txn): balance_a txn.read(account_A) balance_b txn.read(account_B) txn.write(account_A, balance_a - 100) txn.write(account_B, balance_b 100) client.run(transfer) for attempt in range(self.max_retries): # 开启事务 txn self.txn_service.begin() try: # 执行事务逻辑 result txn_func(txn) # 提交 commit_result self.txn_service.commit(txn) if commit_result.success: return result if result is not None else commit_result # 如果需要重试 if commit_result.retry and attempt self.max_retries - 1: logger.info(fRetrying transaction (attempt {attempt 2})) continue raise Exception(fTransaction failed: {commit_result.error}) except Exception as e: self.txn_service.rollback(txn) if attempt self.max_retries - 1: logger.info(fRetrying after error: {e}) continue raise def atomic_update(self, updates: Dict[str, Any]) - bool: 原子批量更新 Example: client.atomic_update({ account_A: 100, account_B: 200 }) def update_func(txn): for key, value in updates.items(): txn.write(key, value) try: self.run(update_func) return True except Exception as e: logger.error(fAtomic update failed: {e}) return False def atomic_read(self, keys: List[str]) - Dict[str, Any]: 原子批量读取 保证读取到一致的数据快照 result {} def read_func(txn): nonlocal result for key in keys: result[key] txn.read(key) self.run(read_func) return result def transfer(self, from_key: str, to_key: str, amount: Any) - bool: 转账操作 原子地从一个key转移到另一个key def transfer_func(txn): balance_from txn.read(from_key) balance_to txn.read(to_key) if balance_from is None: raise ValueError(fSource {from_key} not found) if balance_from amount: raise ValueError(fInsufficient balance: {balance_from} {amount}) txn.write(from_key, balance_from - amount) txn.write(to_key, balance_to amount if balance_to else amount) try: self.run(transfer_func) return True except Exception as e: logger.error(fTransfer failed: {e}) return False六、完整演示# examples/txn_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 from minikv.txn.service import TxnService from minikv.txn.client import TxnClient def demo_basic_transaction(): 演示基本事务操作 print( * 70) print( 分布式事务演示) print( * 70) with tempfile.TemporaryDirectory() as tmpdir: # 启动集群 print(\n 启动集群...) cluster MiniKVCluster( node_count3, base_port9500, data_diros.path.join(tmpdir, kv_data) ) client cluster.start() # 获取Leader节点的事务服务 leader_id cluster.get_leader() leader_service cluster.nodes[leader_id] # 创建事务服务和客户端 txn_service TxnService( leader_service.state_machine, leader_service.raft ) txn_client TxnClient(txn_service) # 初始化账户 print(\n 初始化账户:) client.set(account_A, 1000) client.set(account_B, 500) print(f account_A: {client.get(account_A)}) print(f account_B: {client.get(account_B)}) # 转账事务 print(\n 转账 $300: A → B) success txn_client.transfer(account_A, account_B, 300) print(f 结果: {✅ 成功 if success else ❌ 失败}) # 验证余额 print(f\n 转账后余额:) balances txn_client.atomic_read([account_A, account_B]) for key, value in balances.items(): print(f {key}: {value}) assert balances[account_A] 700 assert balances[account_B] 800 # 原子批量更新 print(\n 原子批量更新:) success txn_client.atomic_update({ account_C: 300, account_D: 600 }) print(f 结果: {✅ 成功 if success else ❌ 失败}) # 验证 accounts txn_client.atomic_read([account_C, account_D]) for key, value in accounts.items(): print(f {key}: {value}) cluster.stop() def demo_concurrent_transactions(): 演示并发事务 print(\n * 70) print(⚡ 并发事务演示) print( * 70) with tempfile.TemporaryDirectory() as tmpdir: cluster MiniKVCluster( node_count3, base_port9600, data_diros.path.join(tmpdir, kv_data) ) client cluster.start() leader_id cluster.get_leader() leader_service cluster.nodes[leader_id] txn_service TxnService( leader_service.state_machine, leader_service.raft ) txn_client TxnClient(txn_service) # 初始化 client.set(counter, 100) print(f\n初始 counter {client.get(counter)}) # 并发递增 import threading def increment(): def inc_func(txn): value txn.read(counter) txn.write(counter, value 1) for _ in range(10): try: txn_client.run(inc_func) except Exception as e: print(f 递增失败: {e}) threads [] for i in range(5): t threading.Thread(targetincrement) threads.append(t) t.start() for t in threads: t.join() final_value client.get(counter) print(f\n 最终 counter {final_value}) print(f 期望值 150 (100 5 * 10)) print(f 结果: {✅ 正确 if final_value 150 else ❌ 错误}) cluster.stop() def demo_transaction_isolation(): 演示事务隔离性 print(\n * 70) print(️ 事务隔离性演示) print( * 70) with tempfile.TemporaryDirectory() as tmpdir: cluster MiniKVCluster( node_count3, base_port9700, data_diros.path.join(tmpdir, kv_data) ) client cluster.start() leader_id cluster.get_leader() leader_service cluster.nodes[leader_id] txn_service TxnService( leader_service.state_machine, leader_service.raft ) txn_client TxnClient(txn_service) # 初始化数据 client.set(x, 10) client.set(y, 20) print(\n初始状态:) print(f x {client.get(x)}) print(f y {client.get(y)}) # 模拟事务隔离 import threading results {} def txn_swap(): 交换x和y的值 def swap_func(txn): x_val txn.read(x) y_val txn.read(y) txn.write(x, y_val) txn.write(y, x_val) try: txn_client.run(swap_func) results[swap] success except Exception as e: results[swap] str(e) def txn_read_both(): 同时读取x和y def read_func(txn): x_val txn.read(x) y_val txn.read(y) results[read_x] x_val results[read_y] y_val # 验证x和y应该来自同一个快照 if x_val y_val: results[anomaly] True try: txn_client.run(read_func) except Exception as e: results[read_error] str(e) # 并发执行 t1 threading.Thread(targettxn_swap) t2 threading.Thread(targettxn_read_both) t1.start() t2.start() t1.join() t2.join() print(\n 隔离性检查:) print(f 交换事务: {results.get(swap)}) print(f 读取 x: {results.get(read_x)}) print(f 读取 y: {results.get(read_y)}) print(f 异常: {⚠️ 发现异常! if results.get(anomaly) else ✅ 正常}) # 最终状态 print(f\n最终状态:) print(f x {client.get(x)}) print(f y {client.get(y)}) cluster.stop() if __name__ __main__: demo_basic_transaction() demo_concurrent_transactions() demo_transaction_isolation()七、测试# tests/test_txn.py import unittest import time import tempfile import os import threading from minikv.txn.types import * from minikv.txn.manager import TxnManager from minikv.txn.service import TxnService from minikv.kv.state_machine import KVStateMachine class TestTxnManager(unittest.TestCase): 事务管理器测试 def setUp(self): self.sm KVStateMachine() self.txn_mgr TxnManager(self.sm) def test_begin_commit(self): 测试开启和提交事务 txn self.txn_mgr.begin() self.assertEqual(txn.status, TxnStatus.PENDING) self.txn_mgr.write(txn, key1, value1) self.txn_mgr.write(txn, key2, value2) result self.txn_mgr.commit(txn) self.assertTrue(result.success) self.assertEqual(result.txn_id, txn.txn_id) def test_rollback(self): 测试回滚 txn self.txn_mgr.begin() self.txn_mgr.write(txn, key, value) self.txn_mgr.rollback(txn) self.assertEqual(txn.status, TxnStatus.ABORTED) self.assertIsNone(self.sm.read(key)) def test_conflict_detection(self): 测试冲突检测 # 先写入一个值 self.sm.apply(type(cmd, (), {operation: SET, key: x, value: 10})()) # 事务1读取 txn1 self.txn_mgr.begin() val1 self.txn_mgr.read(txn1, x) self.assertEqual(val1, 10) # 另一个事务修改 self.sm.apply(type(cmd, (), {operation: SET, key: x, value: 20})()) # 事务1提交应该失败 result self.txn_mgr.commit(txn1) self.assertFalse(result.success) self.assertTrue(result.retry) def test_atomicity(self): 测试原子性部分失败应该全部回滚 txn self.txn_mgr.begin() self.txn_mgr.write(txn, a, 1) self.txn_mgr.write(txn, b, 2) # 模拟提交过程中的失败 # (通过冲突检测触发) self.sm.apply(type(cmd, (), {operation: SET, key: a, value: 999})()) result self.txn_mgr.commit(txn) self.assertFalse(result.success) # 验证数据没有被部分应用 self.assertEqual(self.sm.read(a), 999) # 其他事务的修改 self.assertIsNone(self.sm.read(b)) # 事务的修改被回滚 class TestConcurrentTransactions(unittest.TestCase): 并发事务测试 def test_concurrent_increment(self): 测试并发递增 sm KVStateMachine() txn_mgr TxnManager(sm) # 初始值 sm.apply(type(cmd, (), {operation: SET, key: counter, value: 0})()) def increment(): for _ in range(10): while True: txn txn_mgr.begin() val txn_mgr.read(txn, counter) txn_mgr.write(txn, counter, val 1) result txn_mgr.commit(txn) if result.success: break threads [] for _ in range(5): t threading.Thread(targetincrement) threads.append(t) t.start() for t in threads: t.join() self.assertEqual(sm.read(counter), 50) if __name__ __main__: unittest.main()八、总结这一讲我们为MiniKV实现了分布式事务组件功能事务类型​事务ID、状态、读写集事务管理器​生命周期、冲突检测、OCC事务服务​Raft集成、提交协议事务客户端​自动重试、转账、批量操作关键成果✅ 跨多个key的原子操作✅ 乐观并发控制OCC✅ 自动冲突检测和重试✅ 可串行化隔离级别✅ 事务超时和清理下一讲我们将实现二级索引——让MiniKV支持按非主键字段高效查询。开发之余的小工具推荐​处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。