Python asyncio.timeout 怎样把超时转换成 TimeoutError

在 Python 3.11 及以上，处理 asyncio.timeout() 超时的关键只有一句话：把 except TimeoutError 写在 async with 外面。上下文内部收到的是任务取消信号，退出上下文时才会把它转换成内置的 TimeoutError。

官方文档：https://docs.python.org/3.14/library/asyncio-task.html#asyncio.timeout

最小写法：在上下文外捕获 TimeoutError

asyncio.timeout(delay) 返回一个异步上下文管理器。把真正需要限时的 await 放进去，超时后在外层捕获异常：

import asyncio

async def load_profile():
    # 用 sleep 模拟一个暂时没有返回的外部异步操作
    await asyncio.sleep(2)
    return {"name": "Ada"}

async def main():
    try:
        # 这里只允许当前协程在上下文内等待 0.5 秒
        async with asyncio.timeout(0.5):
            profile = await load_profile()
            print(profile)
    except TimeoutError:
        # 超时转换发生在 async with 退出时，因此这里才能捕获
        print("加载个人资料超时")

asyncio.run(main())

这里的 TimeoutError 是 Python 内置异常，不必写成已经被废弃的 asyncio.TimeoutError。如果把 try/except 放进 async with 内部，通常只能看到 CancelledError 的取消过程，而不是最终的超时异常。

为什么必须在 async with 外面捕获

超时并不是直接从被等待的协程“抛出一个普通异常”。截止时间到达时，asyncio.timeout() 会取消当前任务；上下文管理器拦截这次取消，并在离开上下文时转换为 TimeoutError。因此，捕获位置决定你看到的是取消信号还是可处理的超时结果。

import asyncio

async def call_service():
    try:
        # 被 timeout 取消时，控制流会先进入这里
        await asyncio.sleep(10)
    finally:
        # 无论成功、超时还是外部取消，都应释放临时资源
        print("执行清理逻辑")

async def main():
    try:
        async with asyncio.timeout(1):
            await call_service()
    except TimeoutError:
        # 只有离开 timeout 上下文后，才能统一处理超时
        print("服务调用超过截止时间")

asyncio.run(main())

被包住的协程应该允许取消继续传播，尤其不要在清理完成后无故吞掉 CancelledError。这样超时上下文、任务组等结构化并发工具才能保持自己的控制边界。

动态截止时间：timeout(None)、reschedule 和 expired

如果进入上下文时还不知道时限，可以先传入 None，拿到超时对象后再用事件循环的单调时钟设置绝对截止时间。expired() 用于在上下文退出后判断是否真的越过了截止时间：

import asyncio

async def fetch_data():
    # 模拟获取配置后再决定后续调用的时限
    await asyncio.sleep(0.2)
    return "ready"

async def main():
    timeout_cm = None
    try:
        async with asyncio.timeout(None) as timeout_cm:
            loop = asyncio.get_running_loop()
            # deadline 使用事件循环时钟，不要拿 time.time() 直接相加
            timeout_cm.reschedule(loop.time() + 1.0)
            result = await fetch_data()
            print(result)
    except TimeoutError:
        print("动态截止时间已到")
    finally:
        # timeout_cm 在进入上下文后可用于判断是否越过截止时间
        if timeout_cm is not None and timeout_cm.expired():
            print("记录为超时结果")

asyncio.run(main())

timeout(None) 不表示永远不受控制，而是把设置截止时间的责任延后。多个 timeout 可以安全嵌套；内层先到期时，外层仍然拥有自己的更大范围，但业务代码要避免在多个层级重复记录同一个超时。

asyncio.timeout 和 wait_for 怎么选

两者都能得到 TimeoutError，但边界不同：

wait_for 超时会取消被等待对象，并且可能等待它完成取消，所以总耗时不一定严格等于参数值。若任务不应因等待方超时而取消，需要明确考虑 asyncio.shield()，但这会把任务生命周期交给调用方管理，不能只为“隐藏异常”而使用。

常见误区与使用清单

• 捕获位置：TimeoutError 放在 async with asyncio.timeout(...) 外。
• 版本边界：asyncio.timeout 自 Python 3.11 提供；更旧版本应单独设计兼容方案。
• 取消清理：用 try/finally 关闭连接、文件或临时资源，不要吞掉取消信号。
• 时钟来源：绝对截止时间使用 asyncio.get_running_loop().time()，不要混用墙上时钟。
• 异常范围：超时只代表等待没有在期限内完成，不等于服务返回了业务错误。

一句话总结：用 asyncio.timeout 包住需要限时的异步工作，让取消在内部完成，让 TimeoutError 在外部统一处理；动态场景再配合 reschedule 和 expired。

相关问题

asyncio.timeout 里面能捕获 TimeoutError 吗？通常不能按预期捕获，因为转换发生在上下文管理器退出阶段，应把处理逻辑放在外层。

超时后原来的协程还会继续吗？默认会收到取消请求并停止；如果使用 shield 保留任务，就必须自己保存并管理它的引用和最终结果。