异步 TTL 缓存与击穿锁

功能说明

异步服务经常需要缓存远程查询结果。简单地在协程里先读字典、未命中再请求，会在高并发下产生缓存击穿：多个请求同时未命中，全部打到上游。AsyncTTLCache 在普通 TTL 字典之外增加按 key 的 asyncio.Lock，未命中时只有一个协程执行 loader，其余协程在锁上等待，拿到锁后再次检查缓存，从而复用第一个协程写入的结果。

缓存值以过期时间戳和值组成的元组保存，读取时用注入的单调时钟判断是否过期。时钟可替换，测试时无需真实等待。invalidate 用于主动失效，适合写操作后清理。

关键代码

_fresh 先取出条目，若不存在或时钟已到达过期时间就删除并返回 None，避免返回陈旧值。get_or_load 先做无锁快路径，命中直接返回；未命中才用 setdefault 取得该 key 专属的锁。进入锁后必须再次调用 _fresh，这是双检模式的关键，否则排队协程会重复执行 loader。loader 完成后写入新的过期时间。main 用 gather 同时发起两次读取，并用闭包计数器记录真实加载次数。

使用注意

这里假设缓存值不会为 None，否则命中 None 会被当作未命中；如需缓存 None，应改用哨兵对象或单独的存在标记。_locks 字典会随 key 数量增长，长期运行的服务需要按访问频率清理，或改用固定数量的分段锁。该实现只在单个进程内有效，多进程或多实例部署仍需 Redis 等共享缓存。TTL 从加载完成时开始计算，不是从请求开始时计算。上游失败时异常会传给等待的所有协程，必要时在 loader 内做重试。

验证方式

运行脚本，第一次输出 same: True loader calls: 1，说明两个并发调用只触发一次加载。等待超过 TTL 后再次读取，计数变为 2，证明过期后确实重新加载。可把 ttl 改为更长值并再次运行，观察过期前后计数的差异。

代码示例

import asyncio
import time
from collections.abc import Awaitable, Callable
from typing import Any


class AsyncTTLCache:
    def __init__(self, ttl: float, clock: Callable[[], float] = time.monotonic) -> None:
        self._ttl = ttl
        self._clock = clock
        self._values: dict[str, tuple[float, Any]] = {}
        self._locks: dict[str, asyncio.Lock] = {}

    def _fresh(self, key: str) -> Any | None:
        item = self._values.get(key)
        if item is None:
            return None
        expires_at, value = item
        if self._clock() >= expires_at:
            self._values.pop(key, None)
            return None
        return value

    async def get_or_load(self, key: str, loader: Callable[[], Awaitable[Any]]) -> Any:
        value = self._fresh(key)
        if value is not None:
            return value
        lock = self._locks.setdefault(key, asyncio.Lock())
        async with lock:
            value = self._fresh(key)
            if value is not None:
                return value
            value = await loader()
            self._values[key] = (self._clock() + self._ttl, value)
            return value

    def invalidate(self, key: str) -> None:
        self._values.pop(key, None)


async def main() -> None:
    cache = AsyncTTLCache(ttl=0.2)
    calls = 0

    async def load() -> dict[str, str]:
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.05)
        return {"id": "42", "name": "Ada"}

    first, second = await asyncio.gather(
        cache.get_or_load("user:42", load),
        cache.get_or_load("user:42", load),
    )
    print("same:", first == second, "loader calls:", calls)
    await asyncio.sleep(0.25)
    await cache.get_or_load("user:42", load)
    print("after expiry:", calls)


asyncio.run(main())