返回博客

路径缓存的三个隐患：为什么文件处理要用内容哈希做 key

一个批量 PDF 解析脚本，第一次跑耗时 40 分钟，加了缓存之后理论上重跑应该秒出。 但业务侧整理文件目录时做了一次批量重命名，缓存全部失效，第二次还是跑了 40 分钟—— 没有报错，只是静默地重新算了一遍。

排查过程顺手踩了另外两个坑：序列化崩溃和单测变成了集成测试。 把这三个坑写下来。

现象

| 坑 | 表现 | |----|------| | 路径基缓存 | 文件重命名/移动后缓存 miss，重新处理，无任何报错 | | dataclasses.asdict() 序列化 | 含嵌套 frozen dataclass 的对象写入 JSON 时抛 TypeError | | 缓存逻辑混进处理函数 | 单测 extract_text() 时必须传 cache_dir，否则函数行为不一致 |

根因

坑 1：路径是标识符，不是内容指纹

路径基缓存的隐含假设是「同一路径 = 同一内容」。 只要文件被移动或重命名，这个假设就破了——缓存 key 变了，命中不到，重新处理。 而且破的方式是静默失效，不抛错，很难感知。

# 路径基缓存：rename 之后全 miss
cache = {"/data/reports/2024-q1.pdf": result}
# 文件移到 /data/archive/2024-q1.pdf 之后，key 对不上，白算一遍

内容哈希（SHA-256）解决的是另一个问题：文件身份由内容决定，而不是由位置决定。 重命名 = 同一 hash = 缓存命中。内容改了 = hash 变了 = 自动失效，不需要手动维护任何索引。

坑：文件重命名后路径缓存静默失效

用 path 字符串做 dict key 或文件名，任何 mv/rename/目录整理操作都会导致缓存 miss，且不报错、不提示，只是静默地重新处理了一遍。批处理场景下这意味着白跑几十分钟。

坑 2：dataclasses.asdict() 不擅长嵌套 frozen dataclass

Python 标准库的 dataclasses.asdict() 会递归将嵌套对象转为 dict，但对某些复杂类型（如含 slots=True 的嵌套 frozen dataclass、或自定义类型字段）会抛 TypeError，或者产生意料之外的序列化结构。

# 踩坑写法：看起来省事，实则埋雷
import dataclasses
data = dataclasses.asdict(entry)  # 嵌套类型复杂时 TypeError
json.dumps(data)

asdict() 的递归逻辑依赖 dataclasses.fields()，对非 dataclass 的自定义对象字段直接 copy.deepcopy，序列化行为不可控。手写序列化虽然啰嗦，但完全在掌控范围内。

坑：asdict() + 嵌套 frozen dataclass 在运行时爆炸

dataclasses.asdict() 对含复杂嵌套类型的 frozen dataclass 不保证正确序列化，错误只在运行时出现，静态分析查不到。

坑 3：缓存逻辑混入处理函数，单测变集成测试

# SRP 违反写法：extraction + cache 耦合在一起
def extract_text(path, *, cache_enabled=False, cache_dir=None):
    if cache_enabled:
        # ... 读缓存 ...
    result = _do_extract(path)
    if cache_enabled:
        # ... 写缓存 ...
    return result

这样写的代价：测 extract_text 本身的逻辑，必须关心 cache_dir 是否存在、是否可写。函数有两个职责，测试就得同时覆盖两件事，mock 成本翻倍。

正解

三层拆分：纯处理函数 + 独立缓存读写 + service 层组合。

第一层：内容哈希

import hashlib
from pathlib import Path
 
def compute_file_hash(path: Path) -> str:
    sha256 = hashlib.sha256()
    with open(path, "rb") as f:
        while chunk := f.read(65536):  # 64KB 分块，避免大文件 OOM
            sha256.update(chunk)
    return sha256.hexdigest()

第二层：缓存存储（{hash}.json，O(1) 查找，无需索引文件）

def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None:
    cache_file = cache_dir / f"{file_hash}.json"
    if not cache_file.is_file():
        return None
    try:
        return deserialize_entry(json.loads(cache_file.read_text("utf-8")))
    except (json.JSONDecodeError, ValueError, KeyError):
        return None  # 损坏 = cache miss，不崩溃

第三层：service 层组合（处理函数保持纯粹）

def extract_with_cache(
    file_path: Path,
    *,
    cache_enabled: bool = True,
    cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
    if not cache_enabled:
        return extract_text(file_path)  # 纯函数，完全不知道缓存存在
 
    file_hash = compute_file_hash(file_path)
    cached = read_cache(cache_dir, file_hash)
    if cached is not None:
        return cached.document
 
    doc = extract_text(file_path)
    write_cache(cache_dir, CacheEntry(file_hash, str(file_path), doc))
    return doc
正解：内容哈希 + 服务层分离 + 手写序列化
  • 哈希文件内容（SHA-256，64KB 分块），以 {hash}.json 为文件名存储，O(1) 查找
  • 处理函数保持纯函数，不感知缓存；service 层负责组合
  • 序列化手写，不依赖 dataclasses.asdict()
  • 缓存损坏返回 None（miss），不抛异常，不中断批处理

一句话外卖

缓存 key 应该是内容的指纹，而不是内容的地址——地址会变，指纹不变。