彻底搞懂 functools.lru_cache：原理、用法、坑点与工程化实践

lyjin 2025-09-01

> 关键词：LRU、缓存、性能优化、纯函数、可哈希、装饰器顺序、TTL、异步缓存、工程实践

---

## 为什么要关心 lru_cache？

当你的函数**计算昂贵**、**参数重复**且**没有副作用**时，缓存返回值会显著降低延迟、减少 CPU 消耗。`functools.lru_cache` 是 Python 标准库自带的“轻量级、进程内、函数级缓存”工具——**零依赖，一行生效**。

常见场景：

- 解析/构造对象：正则编译器、分词器、编码器（如 `tiktoken`）、数据库/HTTP 客户端配置等；
- CPU 密集且可复用：复杂数学函数、路径规划子问题、文本分析的子步骤；
- 配置/模板生成：参数 → SQL/Prompt 模板；
- 低频“冷启动”成本高：首次构建词表或模型上下文，之后反复使用。

---

## 它到底做了什么？（原理层面）

**LRU** = *Least Recently Used*（最近最少使用）。当缓存满了，**最久没被用过**的条目会被淘汰。

实现要点（CPython 中的 C 扩展，极快）：

- **哈希表**存“参数签名 → 结果”；
- **双向链表**维护使用顺序（命中时移动到队尾，满了就淘汰队首）；
- **锁**保证并发安全（在 GIL 之外仍有微锁，保护内部结构）；
- 复杂度：**命中/插入摊销 O(1)**；淘汰也近似 O(1)。

> 注意：这是**进程内**缓存。多进程不共享。服务化部署时，请确认你的并发模型。

---

## 一分钟上手

```python
from functools import lru_cache

@lru_cache(maxsize=128)  # 最多缓存 128 条
def heavy(x, y=1):
    print("computing...")  # 命中缓存时看不到这行
    return x ** y

print(heavy(2, 10))  # 第一次计算
print(heavy(2, 10))  # 第二次命中缓存
print(heavy.cache_info())  # CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)
heavy.cache_clear()        # 清空缓存
```

### 常用参数
- `maxsize`: 容量。不给参数等价于 128；`None`（或 `functools.cache`）表示**无限**——谨慎使用，易吃爆内存。
- `typed`: 默认 `False`，即 `1` 与 `1.0` 视为同一键；设为 `True` 则区分类型。

```python
@lru_cache(maxsize=256, typed=True)
def f(x): ...
```

---

## 缓存键如何构造？

- 键由 `*args` + `**kwargs` 规范化组合，要求都**可哈希**：
  - ✅ 可哈希：`int/str/tuple(内部也得可哈希)/frozenset/None`；
  - ❌ 不可哈希：`list/dict/set/自定义可变对象`。

传不可哈希参数会 `TypeError`：把 `list → tuple`、`dict → frozenset(items())` 即可。

```python
from functools import lru_cache

def freeze(obj):
    if isinstance(obj, dict):   return frozenset((k, freeze(v)) for k, v in obj.items())
    if isinstance(obj, list):   return tuple(freeze(i) for i in obj)
    if isinstance(obj, set):    return frozenset(obj)
    return obj  # 假定其余都可哈希

@lru_cache(maxsize=512)
def run(cfg_frozen):
    cfg = dict(cfg_frozen)  # 需要的话再还原
    ...
```

---

## 装饰器顺序的坑（@staticmethod × @lru_cache）

装饰器**自下而上**应用。若写成：

```python
class X:
    @lru_cache(maxsize=32)
    @staticmethod
    def parse(x): ...
```

`lru_cache` 收到的是 `staticmethod` 对象（非可调用），编辑器会警告/运行时报错。**正确顺序**：

```python
class X:
    @staticmethod
    @lru_cache(maxsize=32)  # 先缓存，再变为静态方法
    def parse(x): ...
```

或者把实现放到模块级函数，再 `X.parse = staticmethod(parse_impl)`。

---

## 与类方法/实例方法的相处之道

- 直接装饰实例方法会把 `self` 也纳入键（对象需可哈希），导致**不同实例不共享缓存**；
- 想跨实例共享：用 `@staticmethod`/`@classmethod`，或把函数放到模块级。

```python
class TokenEstimator:
    @staticmethod
    @lru_cache(maxsize=32)
    def encoder(model: str):
        ...
    def estimate(self, text: str, model: str) -> int:
        enc = self.encoder(model)
        return len(enc.encode(text))
```

---

## 适用与不适用（经验之谈）

**非常适合：**
- 纯函数（同参必同果、无副作用）；
- 计算昂贵且参数重复；
- 初始化昂贵但可复用的对象（如 `tiktoken.Encoding`）。

**不太适合：**
- I/O 结果随时间变化（HTTP/DB/文件），除非你接受陈旧数据；
- 有副作用的函数（缓存后副作用就没了）；
- 直接装饰 `async def`（会缓存协程对象，容易“await 两次”错误）。  
  - 异步建议：用第三方 `async_lru`，或把耗时/可缓存的同步部分抽出来。

---

## 性能与并发

- `cache_info()` 可观测命中率；把热点路径命中率提高到 **95%+**，常能带来数量级收益；
- 包装器内部有锁，**线程安全**，但**击穿**（多线程同时 miss）仍可能发生；想“单飞”(single-flight) 需额外协调；
- 进程内缓存，不跨进程。多进程部署可在主进程预热后 `fork`（借助 COW），但不要指望它等同共享缓存。

---

## 进阶：版本号、TTL、分布式

### 1) 版本号进键（配置热切换、词表升级）

```python
from functools import lru_cache

@lru_cache(maxsize=64)
def parse_config(version: int, payload: str):
    ...
# 外部变更时 version += 1，老条目自然被 LRU 淘汰
```

### 2) 近似 TTL（标准库无 TTL）
把**时间片**并入键，例如 5 分钟：

```python
import time
from functools import lru_cache

@lru_cache(maxsize=256)
def fetch_expensive(key: str, slice5m: int):
    return do_fetch(key)

def get(key: str):
    return fetch_expensive(key, int(time.time() // 300))
```

> 每个时间片会新建一组键，上一片数据由 LRU 逐渐淘汰。

### 3) 分布式缓存
跨进程/跨机器：用 Redis 等配合序列化与失效策略；或考虑第三方库：

- `cachetools`: `TTLCache`、`LFUCache`、`LRUCache`、`cachedmethod`（对实例方法很友好）；
- `async_lru`: 给 `async def` 用。

---

## 与 tiktoken 的工程实践（真实范式）

缓存“模型名 → 编码器”能把**初始化成本**摊平到 1 次：

```python
from functools import lru_cache
import tiktoken

@lru_cache(maxsize=32)
def resolve_encoding(model_hint: str):
    name = (model_hint or "").strip().lower()
    try:
        return tiktoken.encoding_for_model(model_hint)
    except Exception:
        base = "o200k_base" if any(tag in name for tag in ("gpt-4o", "gpt-4.1", "o200k")) else "cl100k_base"
        return tiktoken.get_encoding(base)

def estimate_tokens(text: str, model_hint: str = "gpt-4o-mini") -> int:
    s = text or ""
    try:
        enc = resolve_encoding(model_hint)
        return len(enc.encode(s))
    except Exception:
        return max(1, len(s) // 4)
```

若在类里使用：

```python
class ProxyRateLimiter:
    @staticmethod
    @lru_cache(maxsize=32)
    def _resolve_encoding(model_hint: str):
        name = (model_hint or "").strip().lower()
        try:
            return tiktoken.encoding_for_model(model_hint)
        except Exception:
            base = "o200k_base" if any(tag in name for tag in ("gpt-4o", "gpt-4.1", "o200k")) else "cl100k_base"
            return tiktoken.get_encoding(base)

def estimate_tokens(self, text: str, model_hint: str = "gpt-4o-mini") -> int:
        s = text or ""
        try:
            enc = self._resolve_encoding(model_hint)
            return len(enc.encode(s))
        except Exception:
            return max(1, len(s) // 4)
```

---

## 单元测试建议

```python
import time
from functools import lru_cache

def test_basic_hit_miss():
    calls = {"n": 0}
    @lru_cache(maxsize=2)
    def f(x):
        calls["n"] += 1
        return x * x
    assert f(2) == 4     # miss
    assert f(2) == 4     # hit
    assert calls["n"] == 1

def test_typed_flag():
    @lru_cache(maxsize=8, typed=True)
    def f(x): return x
    assert f(1) == 1
    assert f(1.0) == 1.0  # 不同键
    assert f.cache_info().currsize == 2

def test_time_slice_ttl_like():
    @lru_cache(maxsize=8)
    def g(key, slice5m): return time.time()
    t1 = g("a", int(time.time() // 300))
    t2 = g("a", int(time.time() // 300))
    assert t1 == t2       # 同片命中
```

---

## FAQ

**Q: 能缓存异常吗？**  
A: 返回值才会被缓存；异常会穿透，不会缓存。若你希望“异常也缓存（熔断）”，需要自己包装。

**Q: 如何预热（避免首个请求慢）？**  
A: 进程启动时主动调用一遍常见参数，或把热点键入列。

**Q: 多线程下会“同时计算两遍”吗？**  
A: 可能（缓存击穿）。标准库锁只保结构安全，不做 single-flight。要单飞可自建“参数级别”的互斥。

**Q: 装饰 `async def` 可以吗？**  
A: 不建议。会缓存协程对象，`await` 两次会炸。选 `async_lru`，或拆出可缓存的同步子函数。

---

## Cheatsheet（随手查）

- `@lru_cache(maxsize=128, typed=False)`
- `f.cache_info()`／`f.cache_clear()`
- 装饰器顺序：`@staticmethod` **在外层**
- 只缓存**纯函数**；I/O 谨慎
- 避免**无限缓存**，或把**版本号并入键**
- 异步用 `async_lru` 或拆同步子函数

---

> 结语：`lru_cache` 是“低门槛高收益”的优化手段。先找**热点纯函数**下手，用 `cache_info()` 量化命中率，逐步扩大到初始化昂贵的对象/转换逻辑；对动态外部依赖，引入**版本号/时间片**以控制一致性与新鲜度。把它用好，等于给你的服务**免费装了一个一级缓存**。

没有评论

请登陆后评论