Python asyncio.timeout 控制异步操作时限

异步调用最怕“等多久没有边界”。Python 3.11 起，可以用 asyncio.timeout() 把一段 await 包成可取消的超时上下文：任务超过期限时，当前任务会收到取消请求；上下文退出后再把它转换成内置的 TimeoutError。因此，最重要的写法是“在上下文内执行，在上下文外捕获”。

官方地址：https://docs.python.org/3/library/asyncio-task.html

本文只讨论超时上下文、取消与清理，不把超时当作强行终止后台工作的按钮。被取消的协程仍然需要在 finally 中释放自己持有的资源。

先把时限放在真正的等待边界上

一个实用的最小结构如下。上下文只包住需要受这个期限约束的 await，成功时返回结果，超过 2 秒时由外层统一处理。

import asyncio

async def fetch_data():
    # 这里代表一个可能长时间等待的网络或数据库操作
    await asyncio.sleep(3)
    return {"ok": True}

async def main():
    try:
        # 只限制这段等待，不把无关的后续逻辑一起纳入期限
        async with asyncio.timeout(2):
            result = await fetch_data()
            print(result)
    except TimeoutError:
        # 超时转换发生在上下文退出后，所以在这里处理
        print("操作超过时限")

asyncio.run(main())

这里的“取消”作用对象是当前任务，而不是简单地把某个普通函数从线程中杀掉。对于网络请求、队列等待等可中断的异步操作，这种结构能让取消沿着协程栈传播；如果内部捕获了 CancelledError，通常应该完成必要清理后继续抛出。

为什么 TimeoutError 必须在上下文外捕获

asyncio.timeout() 内部先处理 asyncio.CancelledError，在上下文管理器退出时才转换为 TimeoutError。所以把 except TimeoutError 写进 async with 里面，通常接不到这次超时。

import asyncio

async def read_slow_source():
    try:
        # 模拟等待外部资源
        await asyncio.sleep(10)
    finally:
        # 无论成功、超时还是上层取消，都释放本地资源
        print("关闭连接或回收临时状态")

async def main():
    try:
        # TimeoutError 的捕获位置必须在 async with 外层
        async with asyncio.timeout(1):
            await read_slow_source()
    except TimeoutError:
        # 这里得到的是上下文转换后的超时异常
        print("请求超时，返回降级结果")

asyncio.run(main())

清理逻辑不要只放在 except TimeoutError 中，因为调用方也可能主动取消任务。把关闭连接、释放锁、删除半成品等动作放进被取消协程的 finally，才能覆盖这两类退出路径。若确实需要记录取消原因，可以在 except asyncio.CancelledError 中记录后立即 raise。

动态期限和嵌套期限怎么组织

如果创建上下文时还不知道期限，可以先传 None，等拿到预算后用 reschedule() 设置事件循环时钟上的绝对截止点；如果系统已经统一计算了一个截止时间，则直接用 asyncio.timeout_at() 表达更清楚。

import asyncio

async def run_with_budget():
    loop = asyncio.get_running_loop()
    async with asyncio.timeout(None) as scope:
        # 先读取配置或上游预算，再设置绝对截止时间
        scope.reschedule(loop.time() + 2.5)
        await asyncio.sleep(1)
        # expired() 只表示该上下文是否越过了截止时间
        print("是否已经过期:", scope.expired())

async def main():
    try:
        # 外层给整个操作一个总预算，内层可进一步收紧局部预算
        async with asyncio.timeout_at(asyncio.get_running_loop().time() + 5):
            await run_with_budget()
    except TimeoutError:
        # 统一在对应上下文之外处理最终超时
        print("总预算已耗尽")

asyncio.run(main())

嵌套时要先明确预算的归属：外层适合整个请求，内层适合某个数据库或远程服务步骤。不要在多个层级重复吞掉取消，也不要把已经超时的对象继续当作成功结果写入共享状态。

一张表记住选择边界

简而言之：把超时范围缩到真正的等待操作，捕获点放在上下文外，清理写在被取消协程的 finally；期限需要跨层共享时，再使用 timeout_at()。

常见问题

asyncio.timeout 和 asyncio.wait_for 有什么区别？

timeout() 是可嵌套的异步上下文管理器，适合给一段代码建立结构化边界；wait_for() 更像对单个 awaitable 的等待包装。选择时先看你要限制的是一个等待对象，还是一段包含多个 await 的协程逻辑。

捕获 CancelledError 后可以不抛回去吗？

除非你明确设计了取消吞咽和后续状态恢复，否则不建议。通常应在清理后重新抛出，让上层知道任务确实被取消；超时上下文也依赖这条取消链路完成转换。