在线 -
搜索 AIGC编程网
← 返回题目列表

异步代码怎么做缓存？100 个请求同时打同一个 key 会怎样？

中等 第 24 / 27 题 更新于 2026/08/01
异步缓存single-flight缓存击穿并发去重

简化版

异步场景下做缓存有一个同步代码里不明显的问题：await 会让出控制权，所以「检查缓存 → 缓存未命中 → 去查数据库 → 写回缓存」这几步之间，其他协程会插进来。如果 100 个请求同时查同一个 key 而缓存刚好失效，它们会全部发现未命中、全部去查数据库——这就是缓存击穿（也叫 thundering herd / cache stampede）：数据库瞬间承受 100 倍压力，而其中 99 次查询完全是浪费。解决办法叫 single-flight（并发去重）：第一个未命中的请求创建一个 Task 并把它放进「进行中」的字典，后续请求发现已有进行中的 Task 就直接 await 同一个 Task——100 个请求共享 1 次数据库查询。关键点在于「放进字典」这一步必须和「检查」在同一个同步代码块里完成，中间不能有 await，否则又会出现竞态。第二个常见错误是直接用 functools.lru_cache 装饰协程函数：它缓存的是协程对象而不是结果，而协程对象只能被 await 一次——第二次命中缓存时会抛 RuntimeError: cannot reuse already awaited coroutine。第三个要点是异常和 TTL：失败的结果不该被长期缓存（但也要防止「失败 → 立刻重试 → 又失败」的风暴，可以做短 TTL 的 negative caching），而 TTL 到期时的批量失效会造成同时回源（要加随机抖动打散）。核心记忆：异步缓存必须做 single-flight 去重；lru_cache 不能直接用在协程函数上；TTL 要加抖动。

详细版

核心问题与对策：

问题现象对策
缓存击穿热点 key 失效瞬间 N 个请求同时回源single-flight（共享 Task）
缓存雪崩大量 key 同时过期TTL 加随机抖动
缓存穿透查不存在的 key，每次都回源缓存空结果（短 TTL）
lru_cache 装协程cannot reuse already awaited coroutine缓存结果或缓存 Task
慢回源阻塞全部请求所有等待者一起超时stale-while-revalidate
import asyncio, time, random
from typing import Awaitable, Callable

# ① ★反面：没有去重的缓存（缓存击穿）★
_cache: dict[str, tuple[float, object]] = {}

async def bad_get(key):
    hit = _cache.get(key)
    if hit and hit[0] > time.monotonic():
        return hit[1]
    value = await db.query(key)          # ★100 个协程会同时执行到这里★
    _cache[key] = (time.monotonic() + 60, value)
    return value

# ② ★★single-flight：共享同一个 Task★★
_inflight: dict[str, asyncio.Task] = {}

async def get(key: str):
    hit = _cache.get(key)
    if hit and hit[0] > time.monotonic():
        return hit[1]

    task = _inflight.get(key)            # ★① 检查是否已有人在查★
    if task is None:
        # ★② 检查和登记之间不能有 await！（同步代码块，不会被打断）★
        task = asyncio.create_task(_load(key))
        _inflight[key] = task
        task.add_done_callback(lambda _: _inflight.pop(key, None))
    return await task                    # ★③ 所有等待者共享同一个 Task★

async def _load(key):
    value = await db.query(key)
    _cache[key] = (time.monotonic() + 60 + random.uniform(0, 10), value)  # ★抖动★
    return value

# ③ ★lru_cache 的坑★
from functools import lru_cache
@lru_cache(maxsize=128)
async def bad_cached(key):               # ✗ ★缓存的是协程对象★
    return await db.query(key)
# 第二次调用：RuntimeError: cannot reuse already awaited coroutine

# ✓ 变通：缓存 Task（★Task 可以被多次 await★）
@lru_cache(maxsize=128)
def cached_task(key) -> asyncio.Task:
    return asyncio.create_task(db.query(key))
result = await cached_task("k")          # ★可以重复 await 同一个 Task★
# ★但没有 TTL、异常会被永久缓存 → 生产上用专门的库或自己实现★

# ④ 通用的 single-flight 装饰器
def single_flight(fn):
    inflight: dict = {}
    async def wrapper(*args):
        if args in inflight:
            return await inflight[args]           # ★复用★
        task = asyncio.create_task(fn(*args))
        inflight[args] = task
        try:
            return await task
        finally:
            inflight.pop(args, None)              # ★完成后清理★
    return wrapper

# ⑤ ★带 TTL + 负缓存 + 抖动的完整实现★
class AsyncCache:
    def __init__(self, ttl=60, negative_ttl=5, jitter=0.1):
        self._data: dict = {}
        self._inflight: dict[str, asyncio.Task] = {}
        self.ttl, self.negative_ttl, self.jitter = ttl, negative_ttl, jitter

    async def get(self, key, loader: Callable[[], Awaitable]):
        now = time.monotonic()
        entry = self._data.get(key)
        if entry and entry[0] > now:
            return entry[1]
        task = self._inflight.get(key)
        if task is None:
            task = asyncio.create_task(self._load(key, loader))
            self._inflight[key] = task
            task.add_done_callback(lambda _: self._inflight.pop(key, None))
        return await asyncio.shield(task)         # ★shield：调用方取消不影响其他等待者★

    async def _load(self, key, loader):
        try:
            value = await loader()
            ttl = self.ttl if value is not None else self.negative_ttl  # ★空值短 TTL★
        except Exception:
            raise                                  # ★异常不缓存（但会被所有等待者收到）★
        expire = time.monotonic() + ttl * (1 + random.uniform(-self.jitter, self.jitter))
        self._data[key] = (expire, value)          # ★TTL 抖动防雪崩★
        return value

⚠️ 三条必须记住的规则：① 「检查是否有进行中的请求」和「登记自己为进行中」之间绝对不能有 await。因为协程只在 await 处让出控制权，把这两步放在同一个同步代码块里就天然是原子的——这也是异步代码相比多线程的一个优势：不需要锁。反之，如果写成「检查 → await 什么东西 → 登记」，就会有多个协程都通过检查、各自发起查询，去重完全失效。② functools.lru_cache 不能直接装饰 async def——它缓存的是调用结果，而调用一个协程函数得到的是协程对象；协程对象只能被 await 一次，第二次命中缓存时会抛 RuntimeError: cannot reuse already awaited coroutine。变通办法是缓存 Task（Task 可以被多次 await，且天然实现了 single-flight），但要注意它没有 TTL、异常会被永久缓存。③ 所有等待者共享一个 Task 时，取消语义要小心：如果某个等待者被取消（客户端断开），默认情况下 await task 只是取消了它自己的等待；但如果取消传播到了共享的 Task 本身，其他等待者会一起失败——用 asyncio.shield(task) 可以让调用方的取消不影响共享任务。

完整版教学

一、缓存击穿：await 让出控制权带来的竞态

★ 问题的根源：await 是让出点
  async def get(key):
      if key in cache:              # ★① 检查★
          return cache[key]
      value = await db.query(key)   # ★② 让出控制权（几十毫秒）★
      cache[key] = value            # ③ 写回
      return value

  时间线（100 个请求同时到达，缓存刚失效）：
    t0     协程 1 检查缓存 → 未命中 → ★await db.query（让出）★
    t0+ε   协程 2 检查缓存 → ★仍然未命中★（协程 1 还没写回）→ 也去查
    t0+2ε  协程 3 … 一直到协程 100
    t0+50ms 100 个查询陆续返回，各自写回缓存（★99 次是浪费★）

  → ★数据库瞬间承受 100 倍压力★
  → 如果 db.query 本身就慢（比如 500ms），压力窗口更长、堆积更多

★ 三个相关但不同的概念（★面试常混★）：
  ★缓存击穿（hotspot invalid）★
    ★单个热点 key★ 失效瞬间，大量请求同时回源
    → 解法：★single-flight 去重★ / 逻辑过期 / 永不过期 + 后台刷新
  ★缓存雪崩（avalanche）★
    ★大量 key 同时过期★（比如都设了 60 秒且同时写入）→ 集体回源
    → 解法：★TTL 加随机抖动★ / 分批预热
  ★缓存穿透（penetration）★
    查询★根本不存在的 key★ → 每次都回源（缓存永远不命中）
    → 解法：★缓存空结果（短 TTL）★ / 布隆过滤器 / 参数校验

★ 为什么异步下这个问题更突出：
  同步多线程：线程数有限（如 50），最多 50 个并发回源
  ★异步：几千个协程可以同时卡在同一个 await 上★
  → 击穿的放大倍数可能是 ★1000 倍★
  → ★异步服务必须做去重★，不是可选项

★ 一个真实的量化：
  热点商品详情，QPS 2000，缓存 TTL 60 秒，回源耗时 200ms
  无去重：每 60 秒有一个 200ms 的窗口 → ★2000 × 0.2 = 400 个并发回源★
  有去重：★1 个★
  → 数据库连接池（通常 10~50）会被瞬间打满，其他查询全部排队

问题的根源是 await 是让出点：「检查缓存 → 未命中 → await 查库 → 写回」中间的那次 await 会让出控制权几十毫秒，期间其他协程也会检查到未命中并各自发起查询。异步场景下这个问题比同步多线程更突出——多线程受线程数限制（最多几十个并发回源），而几千个协程可以同时卡在同一个 await 上，放大倍数可能是 1000 倍。量化一下：热点商品 QPS 2000、TTL 60 秒、回源 200ms，无去重时每次失效会产生 400 个并发回源（足以打满连接池），有去重则只有 1 个。还要分清三个常被混淆的概念：缓存击穿是单个热点 key 失效瞬间大量回源（解法是 single-flight）、缓存雪崩是大量 key 同时过期（解法是 TTL 抖动）、缓存穿透是查根本不存在的 key（解法是缓存空结果或布隆过滤器）。

二、single-flight：共享同一个 Task

★ 核心思想：让「第一个请求」去干活，其他请求 await 同一个结果

  _inflight: dict[str, asyncio.Task] = {}

  async def get(key):
      # 1. 查缓存
      if (v := cache.get(key)) is not None:
          return v
      # 2. ★检查 + 登记必须在同一个同步块里★
      task = _inflight.get(key)
      if task is None:
          task = asyncio.create_task(load(key))
          _inflight[key] = task
          task.add_done_callback(lambda _: _inflight.pop(key, None))
      # 3. 所有人 await 同一个 Task
      return await task

★ 为什么不需要锁（★异步的优势★）：
  ★协程只在 await 处让出控制权★
  → 从 `_inflight.get(key)` 到 `_inflight[key] = task` 这几行★没有 await★
  → 它们是★原子执行★的，不会被其他协程插入
  ★ 对比多线程：同样的代码需要 threading.Lock 保护

  ✗ 反例（★中间插了 await 就完蛋★）：
    task = _inflight.get(key)
    if task is None:
        await some_check()            # ★让出了！其他协程也会走到这里★
        _inflight[key] = asyncio.create_task(load(key))

★ 用 Task 而不是 Future 的原因：
  ✓ Task 可以被★多次 await★（Future 也可以）
  ✓ Task 会★自动开始执行★（Future 需要有人 set_result）
  ✓ ★异常也会被所有等待者收到★（每个 await 都会抛）
  ★ 但要注意：Task 抛异常时，如果★没有任何人 await★，会有
    "Task exception was never retrieved" 警告 → 所以 done_callback 里
    最好也取一下 exception

★ 清理的两种写法：
  ① done_callback（★推荐★）：
     task.add_done_callback(lambda _: _inflight.pop(key, None))
     → ★任务一结束立刻清理★，包括异常和取消的情况
  ② try/finally：
     try:
         return await task
     finally:
         _inflight.pop(key, None)
     ✗ 问题：★每个等待者都会执行 finally★，第一个执行的就把登记清了，
       后来的请求会重新发起 → ★去重失效★
     → ★所以必须用 done_callback（只在任务完成时执行一次）★

★ 取消的处理（★容易忽略★）：
  100 个请求共享 1 个 Task，其中一个客户端断开了：
    - 它的协程被取消 → `await task` 抛 CancelledError
    - ★共享的 Task 本身不受影响★（其他人继续等）✓
  但如果是这样写：
    task = _inflight[key]
    try:
        return await task
    except asyncio.CancelledError:
        task.cancel()              # ✗ ★把大家的 Task 取消了！★
        raise
  ✓ 更保险：await asyncio.shield(task)
    → ★调用方被取消时，shield 保护内部的 task 继续执行★

single-flight 的核心思想是让第一个请求去干活、其他请求 await 同一个结果。为什么不需要锁是关键洞察：协程只在 await 处让出控制权，所以从 _inflight.get(key) 到 _inflight[key] = task 这几行没有 await，天然是原子执行的——对比多线程里同样的代码需要 threading.Lock 保护。反过来，只要中间插入任何 await，去重就会失效。清理登记有两种写法但只有 done_callback 是对的：用 try/finally 的话每个等待者都会执行 finally，第一个执行的就把登记清掉了，后来的请求会重新发起查询——去重失效；而 done_callback 只在任务完成时执行一次。最后是取消语义：默认情况下某个等待者被取消不影响共享 Task，但如果你在 except CancelledError 里调了 task.cancel()，就会把所有人的 Task 都取消掉——用 await asyncio.shield(task) 更保险。

三、为什么 lru_cache 不能直接用

★ 直接装饰协程函数会怎样：
  @lru_cache(maxsize=128)
  async def get_user(uid):
      return await db.query(uid)

  第一次：get_user(1) → 返回★协程对象 A★ → lru_cache 缓存了 A → await A ✓
  第二次：get_user(1) → ★缓存命中，返回同一个 A★ → await A
         → ★RuntimeError: cannot reuse already awaited coroutine★

  ★ 根本原因：协程对象是"一次性"的
    - 它有内部状态（执行到哪一步了）
    - await 完成后状态是 CLOSED，★不能重新执行★
    → 和生成器只能迭代一次是同一个道理

★ 三种正确做法：
  ① ★缓存 Task 而不是协程★（★顺带实现了 single-flight★）
     @lru_cache(maxsize=128)
     def _get_user_task(uid) -> asyncio.Task:
         return asyncio.create_task(_fetch_user(uid))
     async def get_user(uid):
         return await _get_user_task(uid)      # ★Task 可以多次 await★
     ✓ 优点：简单、天然去重
     ✗ 缺点：★没有 TTL★、★异常会被永久缓存★、
             ★Task 引用被 lru_cache 持有 → 内存泄漏风险★

  ② ★缓存"结果"而不是"可等待对象"★（★最常见★）
     _cache = {}
     async def get_user(uid):
         if uid in _cache:
             return _cache[uid]
         value = await _fetch_user(uid)
         _cache[uid] = value                    # ★缓存最终结果★
         return value
     ✓ 配合 single-flight 和 TTL 才完整

  ③ ★用专门的异步缓存库★
     - aiocache：支持内存/Redis/Memcached 后端，装饰器风格
     - cashews：功能丰富（TTL、锁、early expiration、限流）
     - async-lru：★专门做 lru_cache 的异步版本★
       from async_lru import alru_cache
       @alru_cache(maxsize=128, ttl=60)
       async def get_user(uid): ...

★ 一个易忽略的细节：★lru_cache 用在方法上会内存泄漏★
  @lru_cache
  def method(self, x): ...
  → ★缓存键包含 self → 实例永远不会被回收★
  → 异步场景同理；用 cached_property 或实例级的缓存字典

★ 什么时候可以直接用 lru_cache：
  ✓ ★纯同步的、无 IO 的计算★（在协程里调用也没问题）
    @lru_cache
    def parse_template(s): ...          # ★同步函数，缓存的是结果★
  ✗ 任何 async def 函数

lru_cache 装饰协程函数的问题在于：它缓存的是「调用结果」，而调用协程函数得到的是「协程对象」——协程对象是一次性的（有内部执行状态，await 完成后变成 CLOSED），第二次 await 就抛 RuntimeError: cannot reuse already awaited coroutine（和生成器只能迭代一次是同一个道理）。三种正确做法：① 缓存 Task 而不是协程（Task 可以多次 await，还顺带实现了 single-flight，但没有 TTL、异常会被永久缓存、还有引用泄漏风险）；② 缓存「最终结果」而不是「可等待对象」（最常见，但要配合 single-flight 和 TTL 才完整）；③ 用专门的库（async-lru 的 alru_cache、aiocache、cashews）。还有个易忽略的细节：lru_cache 用在实例方法上会内存泄漏（缓存键包含 self，导致实例永远不被回收）。可以直接用 lru_cache 的只有纯同步无 IO 的计算函数（即使在协程里调用也没问题）。

四、TTL、失效与异常处理

★ TTL 抖动（防雪崩）：
  ✗ 所有 key 都设 ttl=60
    → 如果它们是同时被写入的（比如服务启动预热），★60 秒后同时过期★
    → 集体回源 → 数据库尖峰 → 可能雪崩
  ✓ expire = now + ttl * (1 + random.uniform(-0.1, 0.1))    # ★±10% 抖动★
  ★ 抖动幅度：一般 5%~20%；热点 key 可以更大

★ 异常怎么处理（★三个层次★）：
  ① ★不缓存异常★（默认）
     → 下次请求会重试
     ✗ 风险：下游一直失败 → ★每个请求都去重试 → 打垮下游★
  ② ★负缓存（negative caching）★
     失败结果用★很短的 TTL★缓存（1~5 秒）
     → 既避免了重试风暴，又能快速恢复
  ③ ★熔断★
     连续失败超过阈值 → 直接返回错误/降级值，不再回源

  ★ 注意：single-flight 下异常会传给★所有等待者★
    100 个请求共享一个失败的 Task → 100 个请求都失败
    ✓ 这通常是对的（下游确实挂了），但要确保错误信息合理

★ "空结果"要不要缓存（防穿透）：
  查询不存在的用户 ID → db 返回 None
  ✗ 不缓存 → 每次都查库 → ★恶意请求可以打穿缓存★
  ✓ ★缓存 None，但用更短的 TTL★（如正常 60 秒、空值 5 秒）
  ✓ 大规模场景用★布隆过滤器★先过滤明显不存在的 key

★ 主动失效：
  数据更新时要让缓存失效：
    ① ★删除（推荐）★：cache.pop(key)  → 下次请求重新加载
    ② 更新：cache[key] = new_value    → ★有并发写覆盖的风险★
  ★ 分布式下：本地缓存需要广播失效（Redis pub/sub / 消息队列）
    → ★否则各个实例的本地缓存不一致★

★ stale-while-revalidate（★高级但很实用★）：
  思路：TTL 到期后★先返回旧值★，同时★后台异步刷新★
  → 用户永远不会等待回源（★除了第一次★）
  async def get(key):
      entry = cache.get(key)
      if entry:
          if entry.fresh:
              return entry.value                    # 新鲜 → 直接返回
          if entry.usable:                          # ★过期但还能用★
              spawn(refresh(key))                   # ★后台刷新★
              return entry.value                    # ★立刻返回旧值★
      return await load_and_cache(key)              # 完全没有 → 只能等
  ★ 两个时间：fresh_ttl（新鲜期）< stale_ttl（可用期）
  ★ 代价：可能返回略旧的数据 → ★业务要能接受★

★ 缓存穿透/击穿/雪崩的对策速查：
  ┌──────────┬────────────────────────────────────────┐
  │ 穿透      │ ★缓存空值（短 TTL）★ + 布隆过滤器 + 参数校验│
  │ 击穿      │ ★single-flight★ + 逻辑过期 + 热点永不过期  │
  │ 雪崩      │ ★TTL 抖动★ + 多级缓存 + 限流降级          │
  └──────────┴────────────────────────────────────────┘

TTL 的关键实践是加随机抖动防雪崩（ttl * (1 ± 10%)）——否则同时写入的 key 会同时过期、集体回源。异常处理分三个层次：默认不缓存异常（下次重试，但风险是下游持续失败时每个请求都去重试、把下游打垮）、负缓存（失败结果用 1~5 秒的短 TTL 缓存，既避免重试风暴又能快速恢复）、熔断。要注意 single-flight 下异常会传给所有等待者（这通常是对的）。空结果也要缓存（用更短的 TTL）来防穿透，大规模场景再加布隆过滤器。主动失效推荐「删除」而不是「更新」（后者有并发写覆盖风险），分布式下本地缓存还需要广播失效否则各实例不一致。最后是很实用的 stale-while-revalidate：TTL 到期后先返回旧值、同时后台异步刷新，这样用户永远不会等待回源——代价是可能返回略旧的数据，需要业务能接受。

五、多级缓存与分布式场景

★ 典型的两级结构：
  请求 → ★L1 本地内存缓存★（微秒级）
       → ★L2 Redis★（毫秒级，跨实例共享）
       → 数据库（几十毫秒）

  async def get(key):
      if (v := local.get(key)) is not None:
          return v                                  # L1 命中
      if (v := await redis.get(key)) is not None:
          local.set(key, v, ttl=10)                 # ★回填 L1（短 TTL）★
          return v
      return await single_flight_load(key)          # 回源 + 写两级

★ 各级的 TTL 设计：
  L1（本地）：★短★（5~30 秒）—— 因为无法主动失效，只能靠过期
  L2（Redis）：★长★（5~30 分钟）—— 可以主动删除
  → ★L1 的 TTL 决定了"数据最多有多不一致"★

★ 分布式的 single-flight：
  单机的 _inflight 字典只能去重★本实例内★的并发
  → 10 个实例 = ★10 次回源★（而不是 1 次）
  ✓ 分布式锁：
    if await redis.set(f"lock:{key}", "1", nx=True, ex=10):
        value = await db.query(key)                 # ★只有一个实例查★
        await redis.set(key, value, ex=300)
        await redis.delete(f"lock:{key}")
    else:
        await asyncio.sleep(0.05)                   # ★等一下再读缓存★
        return await get(key)                        # 重试
  ★ 注意：锁的 TTL 要 > 回源耗时；获取不到锁的等待策略要有上限
  ✓ 更简单的替代：★让 L1 的 single-flight + 略微的重复回源可接受★
    （10 个实例回源 10 次，通常比引入分布式锁的复杂度划算）

★ 缓存与数据一致性：
  ★Cache-Aside（旁路缓存，最常用）★
    读：先查缓存 → 未命中查库 → 写缓存
    写：★先更新数据库，再删除缓存★（不是更新缓存）
    ★ 为什么是"删除"而不是"更新"：并发写时更新会互相覆盖出旧值
    ★ 为什么是"先库后缓存"：反过来的话，删缓存后到更新库之间
      有请求进来会把旧值又写回缓存
  ★ 仍然有极小的不一致窗口 → 需要强一致就别用缓存

★ 内存管理（★本地缓存的关键★）：
  ✗ 无界的 dict → ★内存无限增长★
  ✓ 用 LRU 限制条数：
    from collections import OrderedDict
    # 或直接用 cachetools.TTLCache（★线程安全但不是协程感知★，
    #   在单线程事件循环里够用）
    from cachetools import TTLCache
    cache = TTLCache(maxsize=10000, ttl=60)
  ✓ 监控：★命中率、条目数、内存占用★

★ 该不该做本地缓存（★取舍★）：
  ✓ 适合：读多写少、可容忍短暂不一致、热点集中
  ✗ 不适合：★强一致要求★、数据量大（内存放不下）、写频繁
  ★ 一个常被忽略的成本：★多实例的本地缓存让"排查数据不一致"变得很难★

典型的两级结构是「本地内存（微秒）→ Redis（毫秒）→ 数据库（几十毫秒）」，其中 L1 的 TTL 要短（5~30 秒，因为它无法主动失效、只能靠过期，这个 TTL 决定了数据最多有多不一致），L2 可以长且能主动删除。分布式 single-flight 要注意：单机的 _inflight 字典只能去重本实例内的并发——10 个实例就是 10 次回源；要做到全局 1 次需要分布式锁（SET NX EX，注意锁 TTL 要大于回源耗时），但通常「10 个实例回源 10 次」比引入分布式锁的复杂度更划算。一致性方面推荐 Cache-Aside 模式：写操作要「先更新数据库，再删除缓存」——用删除而不是更新是因为并发写更新会互相覆盖出旧值，**用「先库后缓存」**是因为反过来会在窗口期把旧值又写回缓存。最后是内存管理：本地缓存必须有界（用 cachetools.TTLCache 或自己实现 LRU），并监控命中率和条目数。

六、实践清单

★ 实现一个异步缓存的检查清单：
  □ ★single-flight 去重★（检查+登记之间无 await）
  □ ★用 done_callback 清理 inflight★（不能用 try/finally）
  □ ★TTL 加随机抖动★（±10%）
  □ ★空结果也缓存★（短 TTL，防穿透）
  □ ★异常不长期缓存★，但考虑负缓存防重试风暴
  □ ★本地缓存必须有界★（LRU + maxsize）
  □ ★不要用 lru_cache 装 async def★
  □ 共享 Task 时用 ★shield★ 避免一个取消影响所有人
  □ 数据更新时★删除缓存（不是更新）、先库后缓存★
  □ 监控：★命中率、回源 QPS、inflight 数量、缓存条目数★

★ 什么时候不需要这么复杂：
  ✓ QPS 很低（几十）→ 击穿的影响可以忽略
  ✓ 回源很快（<5ms）→ 重复回源的成本低
  ✓ 数据几乎不变 → 启动时加载一次即可（★不用 TTL★）
  ★ 不要过度设计：先测量「重复回源」是不是真的问题

★ 现成的库（★别重复造轮子★）：
  ★async-lru★    alru_cache，最接近 lru_cache 的体验，支持 ttl
  ★aiocache★     多后端（内存/Redis/Memcached）、装饰器、序列化
  ★cashews★     功能最全：TTL、锁（single-flight）、
                early expiration、限流、熔断
  ★cachetools★  同步库，但在单线程事件循环里可以直接用（TTLCache/LRUCache）
  → ★自己实现只在有特殊需求时★

★ 监控指标的解读：
  命中率下降        → TTL 太短 / key 太分散 / 数据在频繁变化
  ★inflight 数量高★ → 回源慢或去重失效
  回源 QPS ≈ 总 QPS → ★缓存基本没起作用★（检查 key 设计）
  内存持续增长      → ★缓存无界或 key 基数太大★

★ 一句话总结：
  ★"异步缓存的核心不是'存起来'，而是'并发未命中时只回源一次'——
    因为 await 会让出控制权，几千个协程可能同时穿透到数据库。
    single-flight（共享 Task）+ TTL 抖动 + 有界内存，
    是三个不能省的要素。"★

实践清单里最关键的四条：single-flight 去重（检查和登记之间无 await）、用 done_callback 而不是 try/finally 清理、TTL 加抖动、本地缓存必须有界。同时要避免过度设计——QPS 很低、回源很快、或数据几乎不变时，这些复杂度都不必要，先测量「重复回源」是不是真的问题。实现上别重复造轮子：async-lru（最接近 lru_cache 的体验）、aiocache（多后端）、cashews（功能最全，自带 single-flight 锁和 early expiration），甚至 cachetools 的 TTLCache 在单线程事件循环里也能直接用。监控指标的解读也很实用：inflight 数量高说明回源慢或去重失效、回源 QPS 接近总 QPS 说明缓存基本没起作用、内存持续增长说明缓存无界或 key 基数太大。

记忆钩子：「异步缓存的核心问题是 ★await 会让出控制权★——『查缓存未命中 → await 回源 → 写回』中间那次 await 期间，其他协程也会检查到未命中并各自回源，这就是★缓存击穿★；而且★异步比多线程更严重★（线程数有限，而几千个协程能同时卡在一个 await 上，放大倍数可达千倍）。★解法是 single-flight：第一个请求创建 Task 并登记到 inflight 字典，后续请求直接 await 同一个 Task★。★关键洞察：检查和登记之间没有 await，所以天然原子、不需要锁★（这是异步相对多线程的优势）——反过来只要中间插入任何 await，去重就失效。★清理 inflight 必须用 done_callback 而不是 try/finally★，因为 finally 会被★每个等待者★执行、第一个执行的就把登记清了。★第二个大坑：lru_cache 不能装饰 async def★——它缓存的是★协程对象★而协程★只能 await 一次★，第二次会抛 cannot reuse already awaited coroutine；变通是★缓存 Task★（可多次 await 且天然去重，但没 TTL、异常永久缓存），或用 async-lru / aiocache / cashews。★三个概念别混★：击穿＝单个热点 key 失效（single-flight）、雪崩＝大量 key 同时过期（★TTL 加 ±10% 抖动★）、穿透＝查不存在的 key（★缓存空值用短 TTL★ + 布隆过滤器）。其他要点：异常默认不缓存但要防重试风暴（★负缓存 1~5 秒★）、共享 Task 时用 ★shield★ 防一个取消影响所有人、★本地缓存必须有界★（LRU+maxsize）、更新数据时★先更新库再删除缓存（删除而非更新，避免并发写覆盖）★、多实例下本地 single-flight 只能去重本实例（★10 个实例仍会回源 10 次★，通常可接受）、★stale-while-revalidate★（过期后先返回旧值 + 后台刷新）能让用户永不等待回源。」

七、常见误区与追问

  • 误区：异步是单线程的，所以缓存读写不会有并发问题。 「不需要锁」和「没有竞态」是两回事。确实单条同步语句序列不会被打断（协程只在 await 处让出），但只要中间有 await，其他协程就会插进来——「检查缓存 → await 回源 → 写回缓存」这个序列里的那次 await 就是竞态窗口：期间到达的所有请求都会发现缓存未命中，然后各自发起回源。而且异步下这个问题比多线程更严重：多线程受线程池大小限制（最多几十个并发回源），而几千个协程可以同时卡在同一个 await 上，热点 key 失效瞬间可能产生上千个并发查询，足以瞬间打满数据库连接池。所以异步服务的缓存必须做 single-flight 去重，这不是可选优化。
  • 误区：@lru_cache 可以直接用来缓存异步函数的结果。 它缓存的是函数调用的返回值，而调用一个 async def 函数返回的是协程对象（此时函数体一行都没执行）。协程对象是一次性的——它带有内部执行状态，被 await 完成后状态变为 CLOSED，第二次 await 会抛 RuntimeError: cannot reuse already awaited coroutine（和生成器只能迭代一次同理）。所以第一次调用「看起来正常」，第二次命中缓存时才炸，很容易在测试里漏掉。三种正确做法：缓存 Task（Task 可以被多次 await，还顺带实现了 single-flight，但没有 TTL、异常会被永久缓存、且 lru_cache 持有 Task 引用有泄漏风险）、缓存最终结果（配合 TTL 和去重）、或直接用 async-lru 的 alru_cache 等成熟库。
  • 误区：清理 inflight 登记用 try/finally 更直观。 这会让去重失效。因为 finally 会在每一个等待者的协程里各执行一次——100 个请求 await 同一个 Task，任务完成后它们依次恢复执行，第一个走到 finally 的就把 _inflight[key] 删掉了；如果此时又有新请求进来，它会发现没有进行中的任务，于是重新发起一次回源。正确做法是 task.add_done_callback(lambda _: _inflight.pop(key, None))——回调只在任务完成时执行一次，与有多少个等待者无关，而且正常完成、异常、被取消三种情况都会触发。顺带一提，在 done_callback 里最好也取一下 task.exception()，避免出现 Task exception was never retrieved 警告。
  • 误区：所有缓存项用统一的 TTL 更简单、更好管理。 统一 TTL 会导致缓存雪崩：如果一批 key 是在同一时刻被写入的（服务启动预热、一次批量刷新、或流量高峰时集中回源），它们就会在同一时刻集体过期，然后同时回源——数据库瞬间承受尖峰，而这个尖峰又会让回源变慢、堆积更多请求，可能反复震荡。解法是给 TTL 加随机抖动：expire = now + ttl * (1 + random.uniform(-0.1, 0.1))，把过期时间打散在一个区间里（幅度一般取 5%~20%，热点 key 可以更大）。这和分布式系统里「重试要加 jitter」是同一个道理——任何会让大量参与者在同一时刻做同一件事的设计，都需要打散。
  • 误区：查询结果为空（数据不存在）时不该写缓存。 不缓存空结果会导致缓存穿透：查询一个不存在的 key，缓存永远不会命中，每次请求都会打到数据库。这在被恶意刷不存在的 ID 时尤其危险——攻击者可以用随机 ID 让缓存完全失效、把数据库打垮。正确做法是缓存空结果，但用明显更短的 TTL（比如正常数据 60 秒、空值 5 秒），这样既挡住了重复查询，又能在数据真的被创建后较快地反映出来。大规模场景还可以在缓存前加布隆过滤器（快速判断「这个 key 肯定不存在」），以及做好参数校验（明显非法的 ID 直接拒绝，根本不查）。
  • 追问：single-flight 里为什么建议用 asyncio.shield？ 因为 100 个请求共享同一个 Task 时，某个等待者被取消不应该影响其他人。典型场景是客户端断开连接，框架取消了对应的 handler 协程——如果你的代码写成 except asyncio.CancelledError: task.cancel(); raise（试图「清理」），就会把大家共享的那个回源 Task 也取消掉，导致另外 99 个还在正常等待的请求全部失败。即使不主动 cancel，某些框架的取消传播也可能波及到 await 的目标。await asyncio.shield(task) 的语义是「保护 task 不受调用方取消的影响」：调用方被取消时，shield 本身会抛 CancelledError，但内部的 task 继续执行，其他等待者不受影响，缓存也能被正常填充。代价是「即使所有等待者都走了，回源仍会跑完」——但这通常正是你想要的（结果会进缓存，下次请求直接命中）。
  • 追问：多实例部署时，single-flight 还有意义吗？ 有，但要理解它的边界。进程内的 _inflight 字典只能去重本实例内的并发：10 个实例同时遇到同一个 key 失效，会产生 10 次回源（而不是 1 次）——但相比「每个实例内部 100 个请求各自回源」的 1000 次，已经降低了 100 倍。大多数场景下这就够了，因为 10 次并发查询数据库完全可以承受。如果确实需要全局只回源一次（回源极其昂贵、或数据库连接池很紧张），可以加分布式锁：用 redis.set(f"lock:{key}", "1", nx=True, ex=10) 抢锁，抢到的实例负责回源并写 Redis，没抢到的短暂 sleep 后重读缓存。但要权衡代价：锁的 TTL 必须大于回源耗时（否则锁提前释放会有多个实例同时回源）、没抢到锁的等待策略要有上限（避免无限重试）、还要处理「持锁实例崩溃」的情况。通常「本地 single-flight + 接受少量重复回源」比引入分布式锁的复杂度更划算。
  • 追问：什么是 stale-while-revalidate，什么时候适合用？ 它的思路是**「TTL 到期后先返回旧值，同时在后台异步刷新」——需要维护两个时间：fresh_ttl（新鲜期，比如 60 秒）和 stale_ttl（可用期，比如 600 秒）。请求到来时：在新鲜期内直接返回；过了新鲜期但还在可用期内，立刻返回旧值并触发一个后台刷新任务；超过可用期或完全没有缓存，才真正等待回源。好处是除了第一次，用户永远不需要等待回源**——P99 延迟会非常稳定，而且天然避免了击穿（后台刷新可以配合 single-flight 只跑一次）。适合的场景是读多写少、数据可以略旧的内容：商品详情、配置、推荐列表、统计数据。不适合强一致要求的场景（账户余额、库存扣减）。实现时要注意：后台刷新任务要保存引用防 GC、要限制并发（避免大量 key 同时触发后台刷新）、失败时不要清掉旧值（保持可用性）。HTTP 的 Cache-Control: stale-while-revalidate 就是这个思想的标准化。

八、加强记忆

异步缓存的核心问题是 await 会让出控制权——「查缓存未命中 → await 回源 → 写回」中间那次 await 期间，其他协程也会检查到未命中并各自回源，这就是缓存击穿；而且异步比多线程更严重（线程数有限，而几千个协程能同时卡在一个 await 上，放大倍数可达千倍，足以瞬间打满数据库连接池）。解法是 single-flight：第一个请求创建 Task 并登记到 inflight 字典，后续请求直接 await 同一个 Task。关键洞察是：检查和登记之间没有 await，所以这段代码天然原子、不需要锁（这是异步相对多线程的优势）——反过来，只要中间插入任何 await，去重就会失效。清理 inflight 必须用 done_callback 而不是 try/finally，因为 finally 会被每个等待者各执行一次、第一个执行的就把登记清掉了。第二个大坑是 lru_cache 不能装饰 async def——它缓存的是协程对象而协程只能被 await 一次，第二次会抛 cannot reuse already awaited coroutine；变通办法是缓存 Task（可多次 await 且天然去重，但没有 TTL、异常会被永久缓存），或直接用 async-lru/aiocache/cashews。三个概念别混：击穿是单个热点 key 失效（用 single-flight）、雪崩是大量 key 同时过期（TTL 加 ±10% 抖动）、穿透是查根本不存在的 key（缓存空值用短 TTL + 布隆过滤器）。其余要点：异常默认不缓存但要防重试风暴（负缓存 1~5 秒）、共享 Task 时用 shield 防一个等待者的取消影响所有人、本地缓存必须有界（LRU + maxsize）、更新数据时先更新库再删除缓存（用删除而非更新，避免并发写互相覆盖）、多实例下本地 single-flight 只能去重本实例内的并发（10 个实例仍会回源 10 次，但通常可接受，全局唯一要靠分布式锁）、stale-while-revalidate（过期后先返回旧值 + 后台刷新）能让用户几乎永不等待回源。