Python asyncio.timeout 怎样把超时变成可捕获异常

在异步服务里，超时最容易出错的地方不是设置秒数，而是捕获位置。asyncio.timeout() 会在上下文管理器内部取消当前任务，再在上下文退出时把这次取消转换成 TimeoutError。因此，想让超时变成可捕获异常，应把 try/except TimeoutError 放在 async with asyncio.timeout(...) 外面。

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

最小正确结构是：try 包住整个 async with，真正可能超时的 await 放进上下文；不要在上下文内部捕获 TimeoutError，因为转换动作要等上下文退出后才完成。

先看值班现场：为什么日志里只有取消，没有超时

我在排查异步接口偶发超时的时候，最初看到的是任务被取消，而不是业务代码期待的超时异常。这个现象并不表示 asyncio.timeout 失效：它的内部机制就是先取消当前 Task，让挂起的 await 尽快收尾，再由上下文管理器把这次取消转成外部的 TimeoutError。

如果代码把捕获写在 async with 里面，异常转换还没有发生，内部通常只能观察到取消过程。排查时先确认三件事：超时上下文是否包住了目标 await，捕获是否位于上下文外部，协程是否真的在等待点把控制权交还给事件循环。

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

下面的结构把限时范围收窄到一次异步调用。TimeoutError 捕获的是超时结果，普通业务异常仍然按原路径向上抛出。

import asyncio


async def fetch_profile():
    # 这里代表需要等待外部资源的异步操作。
    await asyncio.sleep(2)
    return {"name": "Ada"}


async def load_profile():
    try:
        # 只给这次等待设置 0.5 秒的上限。
        async with asyncio.timeout(0.5):
            return await fetch_profile()
    except TimeoutError:
        # 超时转换在上下文退出后完成，所以捕获放在 async with 外部。
        return {"name": "unknown", "source": "timeout-fallback"}


asyncio.run(load_profile())

这里的关键不是返回什么降级值，而是异常边界：fetch_profile() 超过 0.5 秒后，当前任务先收到取消；退出 asyncio.timeout 后，外层 except TimeoutError 才能接住它。

捕获位置怎么判断：内部取消不等于外部超时

不要在限时上下文内部用 except TimeoutError 判断超时。上下文内部发生的是取消，转换要由 asyncio.timeout 的退出逻辑完成。可以把责任边界记成下面这张表：

清理代码不要吞掉取消信号。异步组件依赖取消完成收尾，如果必须捕获 asyncio.CancelledError，通常应在清理完成后继续抛出；否则可能让超时上下文、任务组等结构化并发组件无法正常结束。

值班排查：超时参数生效了，为什么请求仍然很慢

超时只约束事件循环能够观察到的异步等待。它不是把任意同步函数强行变成可中断操作。遇到“设置了 timeout 但请求仍卡住”，按下面顺序判断：

1. 确认目标代码确实在 async with 里，并且没有把 await 写在上下文外。
2. 检查上下文内是否调用了长时间运行的同步函数；同步阻塞会占住事件循环，取消信号只能等它返回后才有机会处理。
3. 检查被调用协程的 finally 是否执行了很慢的异步清理；超时异常的可观察时间还会受到收尾逻辑影响。
4. 区分外部主动取消和超时：请求断开、任务组失败或上层取消，不应一律转换成“服务超时”。

import asyncio


async def call_backend():
    try:
        # 让出事件循环后，timeout 才能在这里交付取消信号。
        await asyncio.sleep(3)
        return "ok"
    finally:
        # 真实代码可在此关闭连接；清理应短小且可重复执行。
        await asyncio.sleep(0)


async def request_once():
    try:
        # 这里的 1 秒只约束 call_backend 这段等待。
        async with asyncio.timeout(1):
            return await call_backend()
    except TimeoutError:
        # 只把确定由 timeout 转换出的异常记为超时。
        return "degraded"
    except asyncio.CancelledError:
        # 上层主动取消要继续传播，避免破坏任务取消语义。
        raise

这个判断可以直接对应告警：TimeoutError 适合统计为下游等待超时；CancelledError 更像调用方撤销、任务组收尾或服务关闭信号，二者的处理和告警级别不应混在一起。

不知道期限时：先创建 None，再 reschedule

有些请求只有读到元数据后才知道真正的等待上限，可以先创建不触发的超时上下文，再用事件循环时钟设置绝对截止时间。

import asyncio


async def load_with_dynamic_deadline():
    loop = asyncio.get_running_loop()
    timeout_cm = None
    try:
        # None 表示先不设置截止时间，便于稍后根据配置计算。
        async with asyncio.timeout(None) as timeout_cm:
            deadline = loop.time() + 1.5
            # reschedule 接收事件循环时钟上的绝对时间。
            timeout_cm.reschedule(deadline)
            return await fetch_profile()
    except TimeoutError:
        # 仍然在上下文外捕获转换后的超时异常。
        return None
    finally:
        # timeout_cm 可用于后续记录是否已越过截止时间。
        if timeout_cm is not None and timeout_cm.expired():
            print("deadline expired")

reschedule() 的参数是绝对时间，不是再追加的秒数；通常用 get_running_loop().time() 计算。需要注意的是，expired() 适合做状态记录，不能替代外层的 TimeoutError 捕获。

回滚路径与复盘清单

如果上线后发现超时比例突然升高，先把降级策略回滚到原有的明确失败路径，保留请求标识和耗时，再逐项检查下面的边界：

• 超时秒数是否覆盖了网络、序列化和清理的合理预算，而不是凭感觉缩短。
• 是否把多个独立下游调用错误地包在一个过大的超时上下文里，导致单个慢调用无法定位。
• 是否在 except TimeoutError 中无条件重试，造成下游压力叠加。
• 是否吞掉了 CancelledError，使服务关闭或上层取消迟迟不能完成。
• 是否需要用 timeout_at() 共享一个绝对截止时间，避免嵌套调用各自重新计时。

复盘时建议同时记录“超时发生在哪个 await、上下文设定的 deadline、是否进入 finally、最终走了降级还是主动取消”。这些信息比单独记录一个“请求超时”更能解释根因。

常见问题

asyncio.timeout 和 asyncio.wait_for 都能超时，怎么选？

asyncio.timeout 是异步上下文管理器，适合给一段代码建立清晰的超时边界，并支持后续查看或调整 deadline；asyncio.wait_for 直接等待一个 awaitable，并在超时后取消它。需要包住多段相邻 await 或动态调整期限时，优先考虑上下文形式。

为什么在 async with 里面捕获不到 TimeoutError？

因为上下文内部先处理的是当前任务取消，TimeoutError 要在上下文退出时才生成。把捕获移到 async with 外部即可。

捕获 TimeoutError 后还要捕获 CancelledError 吗？

如果函数可能被上层取消，建议显式区分并继续传播 CancelledError。不要把所有取消都转换成业务超时，否则服务关闭、客户端断开和任务组失败都会被误记为同一种故障。