题目重点：定义全量快照与覆盖恢复，隔离可变值，验证版本和坏输入，并讨论锁、持久化与增量成本。
题意与假设
这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。
- 训练key为字符串，value为有限JSON树：null、bool、int、有限float、str、list及字符串键dict；不支持对象实例、循环引用和无限深嵌套。
- snapshot返回独立UTF-8 bytes，restore将当前状态完整替换，不做merge；缺失key的get抛KeyError，delete返回是否原本存在。
- set/get复制可变值，调用方不得在方法执行期间并发修改传入对象；本机进程使用锁实现单次方法线性化，不含TTL或跨进程共享。
- 训练无快照文件持久化要求；损坏字节或不支持格式必须失败，失败前后KV状态相同。
解题路线
- 先确认快照是什么：不是暴露当前dict的引用，也不是把所有key分批读完的弱一致视图。它应记录某个逻辑时点的完整状态；之后set/delete不能改变已返回快照。
- 训练选择全量JSON封装{format,data}，格式版本独立于业务值。固定排序键和紧凑分隔符可使同状态字节确定，便于对照；不把确定序列化自动等同校验真实性或完整性。
- set检查类型后深复制，get也返回副本。否则用户保存get结果后修改嵌套list，能绕过锁和API改内部状态；只复制外层dict仍会共享内层容器。snapshot序列化成bytes天然与后续状态隔离。
- snapshot在锁内遍历并编码，序列化过程中没有其他方法写入，因此观察一个完整状态。锁成本随数据量线性增长，且阻塞读写；代码是明确易验证的基础版，不能宣称大数据低延迟。
- restore先在局部解析UTF-8、校验封装字段/格式/JSON类型，拒绝重复对象键和NaN/Infinity，全部通过后持锁一次切换_state。无论中途解析还是验证异常，旧状态都未动。并发写入若在切换前完成，也会被覆盖，这是全量恢复合同的一部分。
- 对更大KV，可在短锁内冻结不可变根或copy-on-write版本，锁外序列化；简单浅复制不能保证嵌套值冻结。增量快照则需基线版本、更新日志、删除tombstone与重放顺序，并确保截断点一致。
- 要持久到磁盘，需要另设计写临时文件、完整性校验、文件同步与原子替换，恢复时验证完整文件再发布；bytes快照本身不能保证进程崩溃后仍存在。异地复制、加密或访问权限也须根据实际部署补合同。
- 测试应在随机set/get/delete/snapshot/restore历史中以事件重放重建期望状态；单独检查调用方修改输入/返回值不影响内部、坏快照不改变当前状态、null键可正确删除。
手动走一遍
设x={items:[1]}，set后修改原list不改变KV；snapshot存下x和y，随后删除y/修改x，restore应把两个key和嵌套值恢复到快照状态。快照缺少y意味着恢复后y必须消失，不能用dict.update误做merge。JSON重复键或错误格式失败后仍保持原状态。
Python 参考实现
下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。
代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。
import copy
import json
import math
import threading
def validate_value(value):
if value is None or type(value) in (bool, int, str):
return
if type(value) is float and math.isfinite(value):
return
if type(value) is list:
for item in value:
validate_value(item)
return
if type(value) is dict and all(type(k) is str for k in value):
for item in value.values():
validate_value(item)
return
raise ValueError('only finite JSON values with string object keys are allowed')
def reject_duplicate_keys(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError('duplicate JSON key')
result[key] = value
return result
class SnapshotKV:
def __init__(self):
self._state = {}
self._lock = threading.RLock()
def set(self, key, value):
if type(key) is not str:
raise ValueError('keys must be strings')
validate_value(value)
owned = copy.deepcopy(value)
with self._lock:
self._state[key] = owned
def get(self, key):
with self._lock:
if key not in self._state:
raise KeyError(key)
return copy.deepcopy(self._state[key])
def delete(self, key):
with self._lock:
if key not in self._state:
return False
del self._state[key]
return True
def snapshot(self):
with self._lock:
text = json.dumps({'format': 'snapshot-kv-v1', 'data': self._state},
ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)
return text.encode('utf-8')
def restore(self, payload):
if type(payload) is not bytes:
raise ValueError('expected UTF-8 bytes')
candidate = json.loads(payload.decode('utf-8'), object_pairs_hook=reject_duplicate_keys)
if type(candidate) is not dict or set(candidate) != {'format', 'data'}:
raise ValueError('invalid snapshot envelope')
if candidate['format'] != 'snapshot-kv-v1' or type(candidate['data']) is not dict:
raise ValueError('unsupported snapshot format')
validate_value(candidate['data'])
with self._lock:
self._state = candidate['data']
复杂度
设总序列化字节B、值树节点数V，snapshot时间O(B+V)，含字节输出O(B)空间；锁持有同量级时间。restore解析/验证O(B+V)，旧状态和新状态在切换前同时存在。set/get为值子树大小线性成本，不是对所有value都O(1)；深度D带来O(D)递归栈限制。
易错点与追问
容易失分的边界
- 返回内部dict或浅复制导致嵌套共享。
- restore用update造成快照不存在的key残留。
- 先清空旧状态再解析，失败丢数据。
- 把内存bytes称为已经持久化的backup。
继续深挖
- 快照期间写入如何降低锁等待？
- 增量快照如何处理删除与顺序？
- 坏文件、旧版本和未知字段怎么迁移？
- 多线程restore是否需要预期当前版本或维护窗口？
