Python asyncio.timeout 和 wait_for 的超时范围怎么选择

如果超时要覆盖一段包含多个 await 的协程逻辑，优先使用 asyncio.timeout()；如果只是给某一个 awaitable 加一层局部等待上限，使用 asyncio.wait_for() 更直观。两者都可能触发取消，但被取消的对象不同：前者取消当前任务在上下文内的执行，后者取消被等待的 awaitable。这个区别会直接影响嵌套调用、异常捕获和清理时间。

先看超时保护的对象是谁

asyncio.timeout(delay) 返回异步上下文管理器。进入后，当前任务在这个上下文内等待的多段工作共享同一条时间边界；超时发生时，内部的 CancelledError 会在上下文外转换成 TimeoutError。因此，捕获超时的 try 应包住 async with，不要只包住其中某一行。

asyncio.wait_for(aw, timeout) 则接收一个 awaitable。传入协程时它会自动调度任务；超时后取消这个被等待的对象，再抛出 TimeoutError。从 API 形状看，它更像给单个调用套一个保护壳，而不是声明整个协程区间的预算。

多段 await 共享总预算时使用 timeout

例如一个请求先取用户，再取权限。若两个调用分别设置 2 秒，整个流程可能拖到 4 秒；如果产品要求“这段准备工作最多 2 秒”，就应该把它们放进同一个超时上下文。

import asyncio

async def load_access_context(user_id):
    # 一个上下文覆盖两次 await，2 秒是这段工作的总预算。
    async with asyncio.timeout(2.0):
        user = await fetch_user(user_id)
        permissions = await fetch_permissions(user["id"])
        return user, permissions

async def handle_request(user_id):
    try:
        return await load_access_context(user_id)
    except TimeoutError:
        # TimeoutError 要在 async with 外捕获，便于统一降级或记录。
        return {"status": "timeout", "user_id": user_id}

这里的边界是“访问上下文准备完成”。后续渲染、写审计日志等工作不应悄悄塞进同一个区间，否则一个慢日志也会消耗业务调用的预算。需要动态 deadline 时，可以先用 asyncio.timeout(None)，拿到上游预算后再通过上下文对象调用 reschedule()。

只限制单个 awaitable 时使用 wait_for

当超时只属于某个独立依赖，wait_for 的表达更贴近意图。下面只限制一次后端查询，查询超时后由调用层决定返回缓存、重试还是失败。

import asyncio

async def query_with_local_timeout(key):
    try:
        # 只给这一项查询设置 800 毫秒上限，不改变外层其它 await。
        return await asyncio.wait_for(query_backend(key), timeout=0.8)
    except TimeoutError:
        # wait_for 已请求取消 query_backend，外层可记录依赖超时。
        return await read_cached_value(key)

async def keep_existing_task(task):
    # shield 只阻止 wait_for 取消 task，不能让 task 获得额外的总预算。
    return await asyncio.wait_for(asyncio.shield(task), timeout=0.8)

shield 要谨慎使用：它把“等待者的超时”与“任务本身是否继续”分开了，任务可能在后台继续占用连接、线程或队列资源。若不需要保留任务，就不要为了绕过取消而加 shield。

把取消、清理和外层预算一起算进去

官方文档特别强调，wait_for 超时后会等待被包装对象真正完成取消；如果被调用协程在 finally 中释放连接、刷写缓冲或等待子任务，调用端观察到的总时间就可能超过 timeout。所以“800 毫秒”更准确的含义是开始取消的时间点，不是所有清理都结束的硬截止线。

生产代码可以按下面的清单落地：

• 需要限制一组连续操作的总时间，用 asyncio.timeout，并把捕获范围放在上下文外。
• 只限制一个依赖调用，用 asyncio.wait_for；记录“开始取消”和“最终返回”两个时间点更容易解释慢请求。
• 外层已有请求 deadline 时，不要在每个子调用随意重新开一条更长预算；局部 timeout 应小于或等于剩余预算。
• 只有明确允许后台继续时才使用 asyncio.shield，并保留任务引用、定义回收和异常记录策略。

常见问题

asyncio.timeout 能替代所有 wait_for 吗？

不能。它更适合包住一个明确的代码区间；对于只想在调用点限制单个 awaitable 的场景，wait_for 更容易阅读，也更容易局部替换。

为什么 timeout 里的 try 捕不到 TimeoutError？

因为上下文内部先收到的是取消信号，timeout 在退出上下文时才把它转换为 TimeoutError。把 try/except 放在 async with 外层即可。

wait_for 设置 1 秒，函数一定 1 秒返回吗？

不一定。超时后它还要等待被包装 awaitable 完成取消；清理逻辑较慢或取消异常时，实际等待可能超过 1 秒。

选择口诀很简单：要限制一段工作的总预算，用 asyncio.timeout；要限制一个 awaitable，用 asyncio.wait_for。再把取消传播、清理耗时和外层 deadline 一起画清楚，超时策略才不会只在正常路径上成立。