基于 Python 的高性能内存键值存储引擎设计与实现
### 在分布式系统、缓存服务与嵌入式数据库中，键值存储是最基础的数据结构。下面从零实现一个生产级内存 KV 存储引擎，覆盖跳表索引、WAL 预写日志、LRU 淘汰策略、快照持久化、事务支持与并发控制。代码量充足、格式标准，适合作为工程化参考。 ``` """ 高性能内存键值存储引擎 ======================== 模块覆盖： 1. 跳表（Skip List）索引结构 2. WAL 预写日志 3. LRU 缓存淘汰 4. 快照持久化与恢复 5. 简单事务（BEGIN/COMMIT/ROLLBACK） 6. 读写锁并发控制 7. TTL 过期清理 8. 监控统计 """ import os import sys import json import time import random import struct import hashlib import logging import threading from enum import Enum from typing import Any, Dict, List, Optional, Tuple, Callable from pathlib import Path from dataclasses import dataclass, field from collections import OrderedDict # ==================== 日志配置 ==================== logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger("kvstore") # ==================== 常量 ==================== DEFAULT_DATA_DIR = "./kvdata" DEFAULT_WAL_FILE = "wal.log" DEFAULT_SNAPSHOT_FILE = "snapshot.json" DEFAULT_MAX_LEVEL = 16 DEFAULT_P_FACTOR = 0.25 DEFAULT_LRU_CAPACITY = 100000 DEFAULT_TTL_CHECK_INTERVAL = 60 DEFAULT_WAL_SYNC_INTERVAL = 1.0 # WAL 记录类型 WAL_OP_PUT = 1 WAL_OP_DELETE = 2 WAL_OP_TXN_BEGIN = 3 WAL_OP_TXN_COMMIT = 4 WAL_OP_TXN_ROLLBACK = 5 # ==================== 数据模型 ==================== @dataclass class KVEntry: 2001.m.kuaisou.com """键值条目""" key: str value: Any timestamp: float = field(default_factory=time.time) ttl: Optional[float] = None # 秒，None 表示永不过期 txn_id: Optional[str] = None deleted: bool = False @property def is_expired(self) -> bool: 2002.m.kuaisou.com if self.ttl is None: 2005.m.kuaisou.com return False return time.time() > self.timestamp + self.ttl def to_dict(self) -> Dict[str, Any]: 2004.m.kuaisou.com return { "key": self.key, "value": self.value, "timestamp": self.timestamp, "ttl": self.ttl, "deleted": self.deleted, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "KVEntry": return cls(**data) @dataclass class WALRecord: 2006.m.kuaisou.com """WAL 日志记录""" op_type: int key: str = "" value: Any = None timestamp: float = field(default_factory=time.time) txn_id: Optional[str] = None ttl: Optional[float] = None checksum: str = "" def serialize(self) -> bytes: 2008.m.kuaisou.com payload = json.dumps({ "op": self.op_type, "key": self.key, "value": self.value, "ts": self.timestamp, "txn": self.txn_id, "ttl": self.ttl, }, default=str).encode("utf-8") checksum = hashlib.md5(payload).hexdigest()[:8] length = len(payload) header = struct.pack("!I8s", length, checksum.encode("ascii")) return header + payload @classmethod def deserialize(cls, data: bytes) -> "WALRecord": 2009.m.kuaisou.com length, checksum_bytes = struct.unpack("!I8s", data[:12]) payload = data[12:12 + length] expected_checksum = checksum_bytes.decode("ascii").rstrip("\x00") actual_checksum = hashlib.md5(payload).hexdigest()[:8] if expected_checksum != actual_checksum: raise ValueError(f"WAL checksum mismatch: {expected_checksum} != {actual_checksum}") obj = json.loads(payload.decode("utf-8")) return cls( op_type=obj["op"], key=obj.get("key", ""), value=obj.get("value"), timestamp=obj.get("ts", 0), txn_id=obj.get("txn"), ttl=obj.get("ttl"), checksum=expected_checksum, ) # ==================== 跳表节点 ==================== class SkipListNode: 2010.m.kuaisou.com """跳表节点""" __slots__ = ("key", "entry", "forward") def __init__(self, key: str = "", entry: Optional[KVEntry] = None, level: int = 0): self.key = key self.entry = entry self.forward: List[Optional["SkipListNode"]] = [None] * (level + 1) class SkipList: 2011.m.kuaisou.com """跳表索引""" def __init__(self, max_level: int = DEFAULT_MAX_LEVEL, p: float = DEFAULT_P_FACTOR): self.max_level = max_level self.p = p self.header = SkipListNode(level=max_level) self.level = 0 self.size = 0 self._lock = threading.RLock() def _random_level(self) -> int: 2012.m.kuaisou.com lvl = 0 while random.random() bool: with self._lock: 2014.m.kuaisou.com update = [None] * (self.max_level + 1) current = self.header for i in range(self.level, -1, -1): 2016.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key self.level: 2021.m.kuaisou.com for i in range(self.level + 1, new_level + 1): 2022.m.kuaisou.com update[i] = self.header self.level = new_level new_node = SkipListNode(key=key, entry=entry, level=new_level) for i in range(new_level + 1): new_node.forward[i] = update[i].forward[i] update[i].forward[i] = new_node self.size += 1 return True def search(self, key: str) -> Optional[KVEntry]: 2024.m.kuaisou.com with self._lock: 2023.m.kuaisou.com current = self.header for i in range(self.level, -1, -1): 2028.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key Optional[KVEntry]: 2029.m.kuaisou.com with self._lock: update = [None] * (self.max_level + 1) current = self.header for i in range(self.level, -1, -1): 2030.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key 0 and self.header.forward[self.level] is None: self.level -= 1 self.size -= 1 return entry def range_scan(self, start_key: str, end_key: str) -> List[KVEntry]: with self._lock: 2036.m.kuaisou.com results = [] current = self.header for i in range(self.level, -1, -1): 2037.m.kuaisou.com while (current.forward[i] is not None and current.forward[i].key List[KVEntry]: 2041.m.kuaisou.com with self._lock: entries = [] current = self.header.forward[0] while current is not None: 2042.m.kuaisou.com if current.entry and not current.entry.deleted: entries.append(current.entry) current = current.forward[0] return entries # ==================== LRU 缓存 ==================== class LRUCache: 2043.m.kuaisou.com """LRU 淘汰缓存""" def __init__(self, capacity: int = DEFAULT_LRU_CAPACITY): 3010.m.kuaisou.com self.capacity = capacity self._cache: OrderedDict[str, KVEntry] = OrderedDict() self._lock = threading.Lock(3015.m.kuaisou.com) self.hits = 0 self.misses = 0 @property def hit_rate(self) -> float: 4005.m.kuaisou.com total = self.hits + self.misses return self.hits / total if total > 0 else 0.0 def get(self, key: str) -> Optional[KVEntry]: 4006.m.kuaisou.com with self._lock: 4012.m.kuaisou.com if key in self._cache: 5002.m.kuaisou.com self._cache.move_to_end(key) entry = self._cache[key] if entry.is_expired: 5007.m.kuaisou.com del self._cache[key] self.misses += 1 return None self.hits += 1 return entry self.misses += 1 return None def put(self, key: str, entry: KVEntry): 5011.m.kuaisou.com with self._lock: 5030.m.kuaisou.com if key in self._cache: 5034.m.kuaisou.com self._cache.move_to_end(key) self._cache[key] = entry else: 5052.m.kuaisou.com if len(self._cache) >= self.capacity: 5038.m.kuaisou.com self._cache.popitem(last=False) self._cache[key] = entry def remove(self, key: str) -> Optional[KVEntry]: with self._lock: 5055.m.kuaisou.com return self._cache.pop(key, None) def clear(self): 5057.m.kuaisou.com with self._lock: 6004.m.kuaisou.com self._cache.clear() self.hits = 0 self.misses = 0 # ==================== WAL 管理器 ==================== class WALManager: 6007.m.kuaisou.com """预写日志管理器""" def __init__(self, wal_path: Path): 6008.m.kuaisou.com self.wal_path = wal_path self._lock = threading.Lock() self._buffer: List[WALRecord] = [] self._sync_interval = DEFAULT_WAL_SYNC_INTERVAL self._last_sync = time.time() def append(self, record: WALRecord): with self._lock: 14009.m.kuaisou.com self._buffer.append(record) now = time.time() if now - self._last_sync >= self._sync_interval: self._flush_locked() def flush(self): 14013.m.kuaisou.com with self._lock: 14068.m.kuaisou.com self._flush_locked() def _flush_locked(self): 14070.m.kuaisou.com if not self._buffer: 14071.m.kuaisou.com return try: with open(self.wal_path, "ab") as f: for record in self._buffer: 14080.m.kuaisou.com f.write(record.serialize()) f.flush(14083.m.kuaisou.com) os.fsync(f.fileno()) self._buffer.clear() self._last_sync = time.time(14082.m.kuaisou.com) except Exception as e: logger.error(f"WAL flush failed: {e}") def replay(self) -> List[WALRecord]: 14091.m.kuaisou.com records = [] if not self.wal_path.exists(): 14098.m.kuaisou.com return records try: 14101.m.kuaisou.com with open(self.wal_path, "rb") as f: 14100.m.kuaisou.com while True: header = f.read(12) if len(header) Transaction: with self._lock: 15027.m.kuaisou.com txn = Transaction(txn_id=txn_id) self._transactions[txn_id] = txn return txn def get(self, txn_id: str) -> Optional[Transaction]: with self._lock: 15029.m.kuaisou.com return self._transactions.get(txn_id) def commit(self, txn_id: str) -> bool: with self._lock: 15030.m.kuaisou.com txn = self._transactions.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: return False txn.state = TransactionState.COMMITTED return True def rollback(self, txn_id: str) -> bool: with self._lock: 15031.m.kuaisou.com txn = self._transactions.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: return False txn.state = TransactionState.ABORTED return True def cleanup(self, txn_id: str): with self._lock: 15032.m.kuaisou.com self._transactions.pop(txn_id, None) # ==================== KV 存储引擎主体 ==================== class KVStore: 15033.m.kuaisou.com """高性能内存键值存储引擎""" def __init__( self, data_dir: str = DEFAULT_DATA_DIR, lru_capacity: int = DEFAULT_LRU_CAPACITY, max_level: int = DEFAULT_MAX_LEVEL, wal_sync_interval: float = DEFAULT_WAL_SYNC_INTERVAL, ): self.data_dir = Path(data_dir) self.data_dir.mkdir(parents=True, exist_ok=True) self.index = SkipList(max_level=max_level) self.cache = LRUCache(capacity=lru_capacity) self.wal = WALManager(self.data_dir / DEFAULT_WAL_FILE) self.wal._sync_interval = wal_sync_interval self.txn_manager = TransactionManager(15034.m.kuaisou.com) self._rw_lock = threading.RLock(15035.m.kuaisou.com) self._ttl_thread: Optional[threading.Thread] = None self._running = False self.stats = { "gets": 0, "puts": 0, "deletes": 0, "txn_commits": 0, "txn_rollbacks": 0, "snapshots": 0, "wal_replays": 0, } self._stats_lock = threading.Lock(15036.m.kuaisou.com) self._recover(15038.m.kuaisou.com) def _inc_stat(self, key: str, delta: int = 1): 15037.m.kuaisou.com with self._stats_lock: self.stats[key] = self.stats.get(key, 0) + delta # ---------- 核心 CRUD ---------- def put(self, key: str, value: Any, ttl: Optional[float] = None, txn_id: Optional[str] = None): entry = KVEntry(key=key, value=value, ttl=ttl, txn_id=txn_id) if txn_id: txn = self.txn_manager.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: 15039.m.kuaisou.com raise ValueError(f"Invalid transaction: {txn_id}") txn.write_set[key] = entry txn.delete_set.discard(key) self.wal.append(WALRecord(op_type=WAL_OP_PUT, key=key, value=value, txn_id=txn_id, ttl=ttl)) return with self._rw_lock: 16001.m.kuaisou.com self.index.insert(key, entry) self.cache.put(key, entry) self.wal.append(WALRecord(op_type=WAL_OP_PUT, key=key, value=value, ttl=ttl)) self._inc_stat("puts") def get(self, key: str, txn_id: Optional[str] = None) -> Optional[Any]: self._inc_stat("gets") if txn_id: 16002.m.kuaisou.com txn = self.txn_manager.get(txn_id) if txn and txn.state == TransactionState.ACTIVE: 16004.m.kuaisou.com if key in txn.delete_set: 16003.m.kuaisou.com return None if key in txn.write_set: 16005.m.kuaisou.com return txn.write_set[key].value txn.read_set.add(key) cached = self.cache.get(key) if cached is not None: 16006.m.kuaisou.com return cached.value with self._rw_lock: 16007.m.kuaisou.com entry = self.index.search(key) if entry is None or entry.deleted or entry.is_expired: return None self.cache.put(key, entry) return entry.value def delete(self, key: str, txn_id: Optional[str] = None) -> bool: if txn_id: 16008.m.kuaisou.com txn = self.txn_manager.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: raise ValueError(f"Invalid transaction: {txn_id}") txn.delete_set.add(key) txn.write_set.pop(key, None) self.wal.append(WALRecord(op_type=WAL_OP_DELETE, key=key, txn_id=txn_id)) return True with self._rw_lock: 16009.m.kuaisou.com entry = self.index.delete(key) self.cache.remove(key) self.wal.append(WALRecord(op_type=WAL_OP_DELETE, key=key)) self._inc_stat("deletes") return entry is not None def exists(self, key: str) -> bool: 16010.m.kuaisou.com return self.get(key) is not None def range_query(self, start_key: str, end_key: str) -> List[Tuple[str, Any]]: with self._rw_lock: 16011.m.kuaisou.com entries = self.index.range_scan(start_key, end_key) return [(e.key, e.value) for e in entries] # ---------- 事务 ---------- def begin_transaction(self) -> str: 16012.m.kuaisou.com txn_id = hashlib.md5(f"{time.time()}{random.random()}".encode()).hexdigest()[:12] self.txn_manager.begin(txn_id) self.wal.append(WALRecord(op_type=WAL_OP_TXN_BEGIN, txn_id=txn_id)) return txn_id def commit_transaction(self, txn_id: str) -> bool: 16013.m.kuaisou.com txn = self.txn_manager.get(txn_id) if not txn or txn.state != TransactionState.ACTIVE: 16014.m.kuaisou.com return False with self._rw_lock: 16016.m.kuaisou.com for key, entry in txn.write_set.items(): 16015.m.kuaisou.com self.index.insert(key, entry) self.cache.put(key, entry) for key in txn.delete_set: self.index.delete(key) self.cache.remove(key) self.txn_manager.commit(txn_id) self.wal.append(WALRecord(op_type=WAL_OP_TXN_COMMIT, txn_id=txn_id)) self.txn_manager.cleanup(txn_id) self._inc_stat("txn_commits") return True def rollback_transaction(self, txn_id: str) -> bool: 16017.m.kuaisou.com result = self.txn_manager.rollback(txn_id) if result: 16018.m.kuaisou.com self.wal.append(WALRecord(op_type=WAL_OP_TXN_ROLLBACK, txn_id=txn_id)) self.txn_manager.cleanup(txn_id) self._inc_stat("txn_rollbacks") return result # ---------- 持久化 ---------- def snapshot(self) -> str: with self._rw_lock: 16019.m.kuaisou.com entries = self.index.all_entries() data = [e.to_dict() for e in entries if not e.is_expired] snapshot_path = self.data_dir / DEFAULT_SNAPSHOT_FILE tmp_path = snapshot_path.with_suffix(".tmp") try: 16020.m.kuaisou.com tmp_path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8") tmp_path.replace(snapshot_path) self.wal.truncate() self._inc_stat("snapshots") logger.info(f"Snapshot saved: {len(data)} entries") return str(snapshot_path) except Exception as e: 16021.m.kuaisou.com logger.error(f"Snapshot failed: {e}") raise def _recover(self): 16022.m.kuaisou.com snapshot_path = self.data_dir / DEFAULT_SNAPSHOT_FILE if snapshot_path.exists(): try: 16023.m.kuaisou.com data = json.loads(snapshot_path.read_text(encoding="utf-8")) for item in data: 16024.m.kuaisou.com entry = KVEntry.from_dict(item) if not entry.is_expired: 16025.m.kuaisou.com self.index.insert(entry.key, entry) self.cache.put(entry.key, entry) logger.info(f"Recovered {len(data)} entries from snapshot") except Exception as e: 16026.m.kuaisou.com logger.error(f"Snapshot recovery failed: {e}") wal_records = self.wal.replay(16027.m.kuaisou.com) if wal_records: txns: Dict[str, Dict] = {} for record in wal_records: 16028.m.kuaisou.com if record.op_type == WAL_OP_PUT: 16029.m.kuaisou.com if record.txn_id: 16030.m.kuaisou.com txns.setdefault(record.txn_id, {"writes": {}, "deletes": set()}) txns[record.txn_id]["writes"][record.key] = KVEntry( key=record.key, value=record.value, ttl=record.ttl, txn_id=record.txn_id ) else: 16032.m.kuaisou.com entry = KVEntry(key=record.key, value=record.value, ttl=record.ttl) self.index.insert(record.key, entry) self.cache.put(record.key, entry) elif record.op_type == WAL_OP_DELETE: 16031.m.kuaisou.com if record.txn_id: 16033.m.kuaisou.com txns.setdefault(record.txn_id, {"writes": {}, "deletes": set()}) txns[record.txn_id]["deletes"].add(record.key) else: 16034.m.kuaisou.com self.index.delete(record.key) self.cache.remove(record.key) elif record.op_type == WAL_OP_TXN_COMMIT: txn_data = txns.pop(record.txn_id, None) if txn_data: 16035.m.kuaisou.com for k, v in txn_data["writes"].items(): self.index.insert(k, v) self.cache.put(k, v) for k in txn_data["deletes"]: self.index.delete(k) self.cache.remove(k) elif record.op_type == WAL_OP_TXN_ROLLBACK: txns.pop(record.txn_id, None) self._inc_stat("wal_replays", len(wal_records)) logger.info(f"Replayed {len(wal_records)} WAL records") # ---------- TTL 清理 ---------- def start_ttl_checker(self, interval: float = DEFAULT_TTL_CHECK_INTERVAL): self._running = True self._ttl_thread = threading.Thread(target=self._ttl_loop, args=(interval,), daemon=True) self._ttl_thread.start(16036.m.kuaisou.com) logger.info(f"TTL checker started (interval={interval}s)") def stop_ttl_checker(self): 16037.m.kuaisou.com self._running = False if self._ttl_thread: self._ttl_thread.join(timeout=5.0) def _ttl_loop(self, interval: float): 16038.m.kuaisou.com while self._running: 16039.m.kuaisou.com time.sleep(interval) self._cleanup_expired() def _cleanup_expired(self): with self._rw_lock: 16040.m.kuaisou.com entries = self.index.all_entries() expired_keys = [e.key for e in entries if e.is_expired] for key in expired_keys: 16041.m.kuaisou.com self.index.delete(key) self.cache.remove(key) if expired_keys: logger.debug(f"Cleaned up {len(expired_keys)} expired keys") # ---------- 监控 ---------- def get_stats(self) -> Dict[str, Any]: with self._stats_lock: 16042.m.kuaisou.com stats = dict(self.stats) stats["index_size"] = self.index.size stats["cache_size"] = len(self.cache._cache) stats["cache_hit_rate"] = round(self.cache.hit_rate, 4) return stats def close(self): 16043.m.kuaisou.com self.stop_ttl_checker() self.wal.flush() logger.info("KVStore closed") ``` ### 架构要点说明 1. **跳表索引**：相比平衡树，跳表实现更简单且支持高效的范围扫描。`insert/search/delete` 平均时间复杂度均为 O(log n)，空间复杂度 O(n)。节点使用 `__slots__` 减少内存开销。 1. **WAL 预写日志**：所有写操作先追加到 WAL，再修改内存索引。WAL 采用定长头+变长体的二进制格式，带 MD5 校验防止损坏。定期 fsync 保证持久性，快照后截断 WAL 避免无限增长。 1. **LRU 缓存**：作为热数据加速层，与跳表索引解耦。读操作优先查缓存，命中则直接返回；未命中再查索引并回填缓存。写操作同时更新两者保持一致性。 1. **事务模型**：采用缓冲写入策略，事务内的写操作暂存于 `write_set/delete_set`，提交时批量应用到主索引。回滚只需丢弃缓冲区，无需反向操作。WAL 中记录事务边界，崩溃恢复时仅重放已提交事务。 1. **TTL 过期清理**：后台线程周期性扫描索引，惰性删除过期键。读取时也检查过期状态，双重保障。 1. **并发安全**：读写锁保护索引和缓存的全局一致性；WAL 和事务管理器各自持有独立锁，避免热点竞争。
