Python 缓存命中后列表却为空：lru_cache 缓存生成器的复现与修复

适用场景与现象

一个导出任务从配置文本读取待处理编号。为了复用解析结果，开发者给解析函数加上 @lru_cache，函数内部用 yield 逐条返回编号。第一次导出正常，第二次相同参数却没有任何数据；重启进程后又暂时恢复。

这种问题容易被当成缓存失效、配置没刷新或下游返回空数据。真正需要检查的是：缓存里保存的究竟是可重复读取的数据，还是已经向前走过的迭代器。

本文针对 Python 3.10 及以上的同步函数、小规模配置快照，不涉及异步生成器、分布式缓存和数据库游标。以下示例没有外部依赖。

一、用十行代码复现

from collections.abc import Iterator
from functools import lru_cache


@lru_cache(maxsize=128)
def iter_ids(source: str) -> Iterator[int]:
    """故障示例：缓存了带有消费进度的生成器对象。"""
    for line in source.splitlines():
        if line.strip():
            yield int(line)


source = "101\n102\n103"
print(list(iter_ids(source)))
print(list(iter_ids(source)))
print(iter_ids.cache_info())

前两行输出分别是 [101, 102, 103] 和 []。缓存统计里可以看到一次 miss、一次 hit。命中率提高了，业务结果却错了。

生成器函数被调用时返回生成器对象，函数体在迭代时才执行。装饰器缓存的是这个对象，并不会把它自动转换成结果列表。第二次调用取回同一个对象，它已经耗尽。如果第一个调用方只读取一个元素，第二个调用方会从第二个元素继续读取，而不是从头开始。

Python functools 官方文档说明，缓存保留参数和返回值的引用，并明确指出生成器不适合直接交给这种结果缓存。这里没有发生缓存数据过期，而是缓存对象的状态已经被消费。

二、现场排查要看返回类型和消费位置

按下面顺序缩小范围：

1. 搜索 @cache、@lru_cache，检查被装饰函数是否包含 yield，或返回 map、filter、iter(...)、数据库游标等一次性迭代对象。
2. 用同一组参数调用两次，并分别完整读取结果；再测试第一次只读取一个元素的情况。
3. 检查 cache_info() 的 hits、misses、currsize。它只能证明缓存行为，不能证明结果正确。
4. 检查是否在日志、校验或计数环节先执行了 list(result)、sum(...)、any(...)，提前消耗了后续还要使用的迭代器。

不要在生产日志里打印完整配置、编号清单或用户数据。记录固定事件、缓存统计、预期数量与实际数量即可。排查时尽量使用独立实例，避免调试读取本身改变共享迭代器的进度。

三、修复：缓存不可变快照，每次独立遍历

对于小规模、可一次性载入内存的配置，把解析结果物化为元组。元组中的元素也使用不可变整数，避免调用方修改缓存内容。将下面代码保存为 id_snapshot.py：

import re
from functools import lru_cache


MAX_SOURCE_CHARS = 4096
ID_PATTERN = re.compile(r"[1-9][0-9]{0,8}")


@lru_cache(maxsize=128)
def _parse_snapshot(source: str) -> tuple[int, ...]:
    """缓存完整成功的解析结果，不保存可消费的迭代器。"""
    values: list[int] = []
    for line_number, raw in enumerate(source.splitlines(), start=1):
        value = raw.strip()
        if not value:
            continue
        if ID_PATTERN.fullmatch(value) is None:
            raise ValueError(f"第 {line_number} 行编号格式错误")
        values.append(int(value))
    return tuple(values)


def load_ids(source: str) -> tuple[int, ...]:
    """读取受大小限制的配置快照；空配置表示没有待处理编号。"""
    if not isinstance(source, str):
        raise TypeError("编号配置必须是字符串")
    if len(source) > MAX_SOURCE_CHARS:
        raise ValueError("编号配置超过字符数上限")
    return _parse_snapshot(source)


def clear_id_cache() -> None:
    """在受控配置切换或测试隔离时清理本进程缓存。"""
    _parse_snapshot.cache_clear()

这里有几个业务约束：

• 配置最多 4096 个字符，入口在进入缓存之前校验；读取文件或接收 HTTP 请求时还应限制原始字节数，不能等整个大文件读完才限制内存。
• 编号范围是 1 到 999999999，只接受 ASCII 十进制且不允许前导零。空行跳过，顺序与重复编号保留。若业务要求去重，需要另行定义规则。
• 缓存键是完整配置文本。内容变化自然对应另一个键；相同文本得到同一不可变快照，不存在仅用文件路径作键后永远读旧内容的问题。
• maxsize=128 限制条目数，并非内存字节数或有效期。需要结合配置规模确定上限；本例不提供 TTL。

调用方可以多次 for item in load_ids(source)。若外部库必须接收迭代器，传入 iter(load_ids(source))，每次创建自己的遍历位置。若需要可修改列表，使用 list(load_ids(source))，修改这个新列表不会污染缓存。

只把 yield 改成返回 list 虽能避免耗尽，却引入共享列表被调用方修改的风险。外层元组也不是通用的深度不可变保证：如果里面放字典或列表，仍会被修改。本例元素是整数，因此可以安全共享。

四、用回归测试覆盖重复读取与部分消费

将下面代码保存为同目录下的 test_id_snapshot.py：

import unittest

from id_snapshot import MAX_SOURCE_CHARS, clear_id_cache, load_ids


class SnapshotTests(unittest.TestCase):
    """验证缓存结果可以重复读取，且错误输入不会被当成空配置。"""

    def setUp(self) -> None:
        clear_id_cache()

    def test_repeated_read(self) -> None:
        source = "101\n102\n103"
        self.assertEqual(list(load_ids(source)), [101, 102, 103])
        self.assertEqual(list(load_ids(source)), [101, 102, 103])

    def test_partial_consumption(self) -> None:
        first = iter(load_ids("101\n102"))
        self.assertEqual(next(first), 101)
        self.assertEqual(list(load_ids("101\n102")), [101, 102])
        self.assertEqual(list(first), [102])

    def test_caller_list_isolation(self) -> None:
        local = list(load_ids("101\n102"))
        local.clear()
        self.assertEqual(load_ids("101\n102"), (101, 102))

    def test_empty_and_order(self) -> None:
        self.assertEqual(load_ids(" \n\n"), ())
        self.assertEqual(load_ids("102\n101\n102"), (102, 101, 102))

    def test_invalid_ids(self) -> None:
        for value in ["0", "01", "-1", "1.5", "１２", "1000000000"]:
            with self.subTest(value=value):
                with self.assertRaisesRegex(ValueError, "第 2 行"):
                    load_ids("101\n" + value)

    def test_input_boundary(self) -> None:
        self.assertEqual(load_ids(" " * MAX_SOURCE_CHARS), ())
        with self.assertRaises(ValueError):
            load_ids(" " * (MAX_SOURCE_CHARS + 1))
        with self.assertRaises(TypeError):
            load_ids(None)  # 故意构造错误类型以验证服务边界。

    def test_changed_source_and_clear(self) -> None:
        self.assertEqual(load_ids("101"), (101,))
        self.assertEqual(load_ids("102"), (102,))
        clear_id_cache()
        self.assertEqual(load_ids("101"), (101,))


if __name__ == "__main__":
    unittest.main()

运行：

python -m unittest -v test_id_snapshot

应执行 7 项测试，不能只检查进程退出码或接受“运行 0 项”。最关键的两项是重复完整读取、部分消费后独立读取，断言必须比较完整编号与顺序，单独检查“函数没有抛异常”无法识别静默漏数据。

验证回归检测能力时，可临时把 _parse_snapshot 的 return tuple(values) 改成 return iter(values)，重跑测试。重复读取和部分消费用例应失败；随后恢复元组实现并重新跑全套。这个故障变体保留了输入校验，只改变缓存对象类型，能直接检验测试是否守住了修复点。

本次在 Windows、Python 3.14.6 上执行上述 7 项 unittest，全部通过。替换为迭代器故障变体后，6 项失败、无执行错误；恢复修复后 7 项再次通过。正文三段 Python 代码语法检查通过。环境未安装 Ruff，未执行 Ruff 格式与静态检查，也未新增依赖。

五、上线与回退的边界

先在预发布环境对同一配置执行两次完整导出，再并行触发两个只读消费方，对比业务编号集合、顺序和行数。发布前确认调用方是否依赖 next(load_ids(...))；返回类型改成元组后，这类调用必须改成 next(iter(load_ids(...)))。类型标注和接口说明也要同步更新。

紧急止损可以先撤掉生成器函数上的缓存，让每次调用产生独立生成器，并评估重新解析带来的负载。清空旧缓存只能让下一次暂时成功，不能修复共享迭代器。应保留最近一次正确导出的对账依据，按业务幂等规则补发漏掉的数据，不要盲目重复执行有副作用的任务。

对于百万行导出，不要为了使用缓存而整体转成元组，应每次新建流并独立管理文件、游标和连接生命周期。lru_cache 是进程内缓存，多 worker 各自保存一份；其内部结构线程安全，也不保证相同键的首次计算在并发下只执行一次。不要给包含写库、扣款等副作用的函数套用本例。

本文的验证范围是本地同步解析和调用方遍历语义，不覆盖真实文件更新、Web worker、生产权限、并发首次计算、内存压力或跨进程失效。这些边界应在接入实际服务时另外验证。

总结

缓存命中不等于数据可复用。生成器带有消费进度，缓存它会把进度共享给后续调用者。小数据使用不可变完整快照，调用方各自创建迭代器；大数据保留独立流式读取。回归测试必须覆盖第二次读取和中途消费，才能发现“没有报错但少了数据”的问题。