何时使用
适用:
- 构建文件处理流水线(PDF 解析、OCR、文本抽取、图像分析),处理成本高且同一文件会被反复处理。
- CLI 工具需要
--cache/--no-cache开关,按内容跨次运行复用结果。 - 批处理中同一文件在多个批次反复出现。
- 想给一个已有的纯处理函数加缓存,但不愿改动它的签名与职责。
不该用(负边界):
- 数据必须始终最新(实时行情、消息流等)——缓存会返回过期结果。
- 单条缓存条目极大——
{hash}.json全量读写会吃内存,改用流式/分块存储。 - 结果不只取决于文件内容,还依赖内容之外的参数(如不同的抽取配置、模型版本、语言选项)——纯内容哈希会错误命中;需把这些参数一并并入缓存键。
- 文件内容本身高频变动,几乎每次都未命中——缓存收益为负。
步骤 / 指令
1. 用内容(非路径)算缓存键:对文件内容做 SHA-256,大文件分块读取,避免整文件入内存。
2. 缓存条目用 frozen dataclass 表示(file_hash + source_path + 处理结果)。
3. 按 {hash}.json 落盘:O(1) 按哈希查,无需索引文件;首次写入时惰性建目录。
4. 用独立服务层包裹纯处理函数:服务层做「查缓存 -> 未命中则处理 -> 写缓存」,纯函数对缓存零感知(SRP)。
5. 读缓存时优雅降级:JSON 损坏/反序列化失败一律当未命中返回 None,不崩溃,下次运行自动重算。
6. 用截断哈希(hash[:12])记录命中/未命中日志,便于排查。
核心约束:
- 哈希内容而非路径——路径会变,内容身份不变。改名/移动 = 缓存命中;内容改动 = 自动失效;无需任何索引文件。
- 处理函数保持纯净,不得在其内部塞入缓存逻辑(否则一函数两职责,违反 SRP)。
- 嵌套 frozen dataclass 不要用
dataclasses.asdict()(复杂嵌套类型会出问题),改用手写序列化以完全掌控格式。
示例
- 内容哈希缓存键(大文件分块):
import hashlib
from pathlib import Path
_HASH_CHUNK_SIZE = 65536 # 64KB 分块,适配大文件
def compute_file_hash(path: Path) -> str:
"""文件内容的 SHA-256(分块读取)。"""
if not path.is_file():
raise FileNotFoundError(f"File not found: {path}")
sha256 = hashlib.sha256()
with open(path, "rb") as f:
while chunk := f.read(_HASH_CHUNK_SIZE):
sha256.update(chunk)
return sha256.hexdigest()
- 缓存条目(frozen dataclass)+ 文件存储:
from dataclasses import dataclass
import json
@dataclass(frozen=True, slots=True)
class CacheEntry:
file_hash: str
source_path: str
document: "ExtractedDocument" # 被缓存的处理结果
def write_cache(cache_dir: Path, entry: CacheEntry) -> None:
cache_dir.mkdir(parents=True, exist_ok=True) # 惰性建目录
cache_file = cache_dir / f"{entry.file_hash}.json"
data = serialize_entry(entry) # 手写序列化
cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
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:
data = json.loads(cache_file.read_text(encoding="utf-8"))
return deserialize_entry(data)
except (json.JSONDecodeError, ValueError, KeyError):
return None # 损坏当未命中,下次重算
- 服务层包裹纯函数(保持 SRP):
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:
logger.info("Cache hit: %s (hash=%s)", file_path.name, file_hash[:12])
return cached.document
logger.info("Cache miss: %s (hash=%s)", file_path.name, file_hash[:12])
doc = extract_text(file_path) # 纯函数
write_cache(cache_dir, CacheEntry(file_hash, str(file_path), doc))
return doc
反模式(避免):
# 坏:路径作键 —— 文件移动/改名即失效
cache = {"/path/to/file.pdf": result}
# 坏:把缓存逻辑塞进处理函数 —— 一函数两职责,违反 SRP
def extract_text(path, *, cache_enabled=False, cache_dir=None):
if cache_enabled: ...
# 坏:嵌套 frozen dataclass 用 asdict() —— 复杂嵌套类型会出错,改用手写序列化
data = dataclasses.asdict(entry)
注意事项
- 哈希内容不哈希路径;大文件务必分块,别整文件读入内存。
- 处理函数保持纯净——缓存是独立关注点,放服务层。
- 缓存损坏/反序列化失败优雅降级为未命中(返回
None),绝不崩溃。 - 命中/未命中均记日志(用截断哈希)便于调试。
- 关键设计取舍:SHA-256 内容哈希=路径无关+内容变即失效;
{hash}.json命名=O(1) 查、免索引;手写 JSON 序列化=完全掌控 frozen dataclass 的格式;mkdir(parents=True)=首写惰性建目录。 - 若结果依赖内容外参数(抽取配置/模型版本),把这些参数一并并入缓存键,否则会错误命中。
互见
- related:
file-upload-storage—— 文件上传与存储管理,缓存层常落在其下游处理环节。 - related:
python-performance-optimization—— 缓存是消除重复高成本计算的性能手段之一。 - related:
clean-code-principles—— 服务层包裹纯函数即单一职责(SRP)的具体落地。 - combines_with:
bullmq-job-queue—— 在批处理/后台队列中复用内容哈希缓存,跳过已处理文件。
采编自 affaan-m/everything-claude-code(MIT 许可)。