实战题解：带 TTL 过期与嵌套事务的 In-Memory Key-Value Store

在北美科技面试中，In-Memory Key-Value Store（内存键值存储系统） 是最具标志性的 Practical Coding 考题之一。从 AI Infra 公司（如 Baseten）到大厂（如 Uber、Airbnb、Rippling），面试官通常会分阶段递进考察：

• Part 1：基本的 SET, GET, DELETE
• Part 2：支持基于时间戳的 SET_WITH_TTL 与惰性/主动过期淘汰
• Part 3：支持嵌套事务（BEGIN, COMMIT, ROLLBACK）与回滚机制

---

核心架构设计

graph TD
    A[Transactional KV Store] --> B[全局主数据字典 main_store]
    A --> C[事务栈 transaction_stack]
    C --> C1[栈帧 1: 当前事务变更集 Delta]
    C --> C2[栈帧 2: 嵌套子事务变更集 Delta]
    A --> D[TTL 堆 expire_heap]

---

工业级标准实现（Python）

import time
import heapq
from typing import Optional, Dict, List

class Tombstone:
    """表示在事务中被删除的标记"""
    pass

class TransactionalKVStore:
    def __init__(self):
        self.global_store: Dict[str, str] = {}
        self.ttl_store: Dict[str, float] = {}  # key -> expire_at_timestamp
        self.ttl_heap: List[tuple] = []  # (expire_at_timestamp, key)
        self.transactions: List[Dict[str, object]] = []

    def _clean_expired(self, current_time: Optional[float] = None):
        """惰性结合最小堆淘汰过期 Key"""
        now = current_time if current_time is not None else time.time()
        while self.ttl_heap and self.ttl_heap[0][0] <= now:
            expire_at, key = heapq.heappop(self.ttl_heap)
            if self.ttl_store.get(key) == expire_at:
                del self.ttl_store[key]
                self.global_store.pop(key, None)

    def set(self, key: str, value: str, ttl_seconds: Optional[float] = None) -> None:
        self._clean_expired()
        
        # 如果处于事务中，写入当前最顶层事务栈帧
        if self.transactions:
            self.transactions[-1][key] = value
        else:
            self.global_store[key] = value

        # 处理 TTL
        if ttl_seconds is not None:
            expire_at = time.time() + ttl_seconds
            self.ttl_store[key] = expire_at
            heapq.heappush(self.ttl_heap, (expire_at, key))
        elif key in self.ttl_store:
            del self.ttl_store[key]

    def get(self, key: str) -> Optional[str]:
        self._clean_expired()

        # 1. 检查事务栈（由顶向下查找）
        for tx in reversed(self.transactions):
            if key in tx:
                val = tx[key]
                return None if isinstance(val, Tombstone) else str(val)

        # 2. 检查全局存储
        if key in self.ttl_store and time.time() > self.ttl_store[key]:
            return None

        return self.global_store.get(key, None)

    def delete(self, key: str) -> bool:
        self._clean_expired()
        
        if self.get(key) is None:
            return False

        if self.transactions:
            self.transactions[-1][key] = Tombstone()
        else:
            self.global_store.pop(key, None)
            self.ttl_store.pop(key, None)
        return True

    def begin(self) -> int:
        """开启新事务，返回当前嵌套深度"""
        self.transactions.append({})
        return len(self.transactions)

    def commit(self) -> bool:
        """提交最顶层事务"""
        if not self.transactions:
            return False
        
        current_tx = self.transactions.pop()
        
        # 如果还有外层事务，合并到外层事务
        if self.transactions:
            self.transactions[-1].update(current_tx)
        else:
            # 提交到全局主存储
            for k, v in current_tx.items():
                if isinstance(v, Tombstone):
                    self.global_store.pop(k, None)
                    self.ttl_store.pop(k, None)
                else:
                    self.global_store[k] = str(v)
        return True

    def rollback(self) -> bool:
        """放弃当前最顶层事务的全部修改"""
        if not self.transactions:
            return False
        self.transactions.pop()
        return True

---

核心考点与性能优化

1. 写时复制（Copy-on-Write）vs 差异标记（Delta Log）：
   • 很多候选人会在 begin() 时把整个全局字典 copy.deepcopy()，这在数据量大时会导致 $O(N)$ 灾难性开销。
   • 最佳实践：采用上述 Delta 栈设计，每次 begin() 只创建一个空字典，开销仅为 $O(1)$。
2. Tombstone 墓碑标记机制：
   • 在事务内调用 delete(key) 时，不能直接在全局删，必须在当前事务记录一个特殊的 Tombstone()，以覆盖外层事务或全局的值。

---

相关实战资源

• In-Memory File System 实战全攻略
• 北美高薪 FinTech & SaaS 独角兽 Practical Coding 考法盘点
• SDE 1 对 1 定制面试辅导