Python asyncio.timeout 区分超时与主动取消的边界

在异步服务里，“这次操作超时了”和“上游已经不需要这次操作”经常都会表现为取消，但它们不是同一个结果。asyncio.timeout() 到期时会取消当前任务，并在上下文管理器外把这次超时转换成 TimeoutError；如果调用方主动取消任务，外层通常应该保留 CancelledError，让取消继续向上传播。

实用判断是：只在 async with asyncio.timeout(...) 外捕获 TimeoutError，资源释放统一放在 finally，不要把所有 CancelledError 都改写成“超时”。本文只围绕这条边界展开。

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

超时是截止时间触发的控制流转换，主动取消是调用方发出的生命周期信号。前者可以在边界处转成业务结果，后者通常必须重新抛出。

先把三种结果放到同一张表里

一个可等待操作至少有三种结局。代码结构应先区分结局，再决定是否重试、降级或退出。

这里的关键不是异常名称，而是信号来源。TimeoutError 表示某个 timeout 上下文确认截止时间已过；CancelledError 仍表示任务生命周期被外部打断。Python 文档还指出，CancelledError 属于 BaseException，普通的 except Exception 不会捕获它。

asyncio.timeout 到期时到底发生了什么

asyncio.timeout(delay) 是异步上下文管理器。到达相对秒数后，它会取消正在执行上下文的当前任务；上下文管理器在退出时处理内部的 asyncio.CancelledError，并把它转换为内置的 TimeoutError。所以 TimeoutError 的捕获位置必须在 async with 外部。

import asyncio


async def read_remote() -> bytes:
    # 这里代表一次可能长时间等待的网络或文件操作。
    await asyncio.sleep(2)
    return b"payload"


async def read_with_deadline() -> bytes | None:
    try:
        # 超时上下文负责给当前任务设置等待截止时间。
        async with asyncio.timeout(0.5):
            return await read_remote()
    except TimeoutError:
        # 只有离开 async with 后，才能在这里接住超时转换结果。
        print("operation timed out")
        return None


asyncio.run(read_with_deadline())

不要把 except TimeoutError 塞进 async with 内部期待它生效。超时发生时，内部首先感知的是取消；转换动作由上下文退出逻辑完成，外层才是稳定的处理边界。

资源清理要跟着被取消的操作走

超时并不等于所有底层资源自动恢复。持有连接、响应对象、临时锁或中间文件的协程，应把释放动作放在自己持有资源的那一层 finally 中。即使只打算让超时在外层变成降级结果，内部也不能跳过清理。

async def fetch_body(client, url: str) -> bytes:
    response = None
    try:
        # 请求返回后才持有 response，清理逻辑因此需要允许它为空。
        response = await client.get(url)
        return await response.read()
    except asyncio.CancelledError:
        # 主动取消和超时取消都要先让资源清理完成，再传播控制流。
        raise
    finally:
        # 伪代码表示归还连接或关闭响应，具体方法取决于客户端库。
        if response is not None:
            await response.close()

如果底层库的关闭方法不是异步方法，应按库的真实接口调用，不要为了“统一写法”强行使用 await。这里真正重要的是：清理属于资源拥有者，且捕获 CancelledError 后不要直接返回一个成功值。

怎样把主动取消和超时结果分开

主动取消通常发生在另一个任务持有当前任务对象，并调用 cancel()。下面的包装函数把两种信号放到不同分支：超时可以转成一个业务状态，主动取消则清理后继续抛出。

import asyncio
from dataclasses import dataclass


@dataclass
class Outcome:
    # 用明确状态区分成功与可接受的超时降级。
    value: bytes | None
    timed_out: bool = False


async def fetch_with_boundary(client, url: str, seconds: float) -> Outcome:
    try:
        # 这里只负责“截止时间到了”的那条路径。
        async with asyncio.timeout(seconds):
            value = await fetch_body(client, url)
            return Outcome(value=value)
    except TimeoutError:
        # 超时是可被业务层观察的结果，例如返回缓存或稍后重试。
        return Outcome(value=None, timed_out=True)
    except asyncio.CancelledError:
        # 调用方主动取消不能伪装成超时，否则上游无法停止等待。
        raise


async def caller(client) -> None:
    task = asyncio.create_task(fetch_with_boundary(client, "https://api.example/", 5))
    try:
        # 业务条件改变后，调用方主动撤销这次工作。
        await asyncio.sleep(0.1)
        task.cancel()
        await task
    except asyncio.CancelledError:
        # 这里收到的是调用方取消语义，应该通知更上层并停止后续工作。
        print("caller cancelled the fetch")

示例中的地址只是代码里的占位字符串，不是本文要介绍的具体服务入口。真实项目中应替换为业务客户端的目标地址，并为取消后的日志记录调用方、任务标识和清理结果，避免把正常的用户离开、请求断开和服务故障混在同一类“超时”指标里。

动态截止时间用 reschedule，不要重复包裹任务

有些请求在进入上下文时还不知道最终预算，例如先读取元数据，再根据元数据决定正文读取的剩余时间。此时可以用 asyncio.timeout(None) 创建未设置截止时间的上下文，再通过返回的 Timeout 对象调用 reschedule()。截止时间使用事件循环的单调时钟。

async def fetch_after_metadata(client, url: str) -> bytes:
    loop = asyncio.get_running_loop()
    try:
        # 先创建没有截止时间的上下文，等待后续预算确定。
        async with asyncio.timeout(None) as scope:
            budget = await client.read_budget()
            # reschedule 接收 loop.time() 体系下的绝对时间点。
            scope.reschedule(loop.time() + budget)
            return await fetch_body(client, url)
    except TimeoutError:
        # 动态截止时间过期后，仍在上下文外处理超时。
        raise RuntimeError("remote body exceeded its budget")

如果只想在上下文结束后判断是否越过截止时间，可以检查 scope.expired()。它适合写入诊断字段，但不能代替异常处理：操作超时仍然需要由外层决定是降级、重试还是失败。

嵌套 timeout 时按最近的业务边界收口

超时上下文可以安全嵌套。常见做法是外层给整次请求设置总预算，内层给某个依赖设置更短预算。内层超时先在自己的边界外转换成 TimeoutError，外层再决定是否继续使用剩余预算。不要把内外层的超时都记成同一个指标名。

async def load_page(client) -> bytes | None:
    try:
        # 外层预算覆盖整个页面加载过程。
        async with asyncio.timeout(3):
            try:
                # 内层预算只约束一个较慢的推荐接口。
                async with asyncio.timeout(0.8):
                    return await fetch_body(client, "/recommendations")
            except TimeoutError:
                # 内层超时可以转成可选数据缺失，继续页面主流程。
                return b""
    except TimeoutError:
        # 外层超时表示整个页面预算不足，和推荐接口超时分开记录。
        return None

这里的业务选择不是 asyncio 自动替你做出的：内层超时后是否继续、是否消耗剩余预算、是否把空值视为降级，都应该由调用方明确规定。

发布前的快速判断清单

• 只在 async with asyncio.timeout() 外捕获 TimeoutError。
• 在持有资源的协程内部用 finally 释放资源，必要时在 except asyncio.CancelledError 中记录后重新抛出。
• 调用方通过 Task.cancel() 主动撤销时，不要把 CancelledError 改写成“超时”。
• 使用 reschedule() 时传入事件循环时钟体系下的绝对截止时间，并在上下文外处理异常。
• 嵌套超时按业务边界命名指标，让“依赖慢”和“总预算耗尽”可以分别定位。

如果只记住一个实现原则，可以把它写成：超时是某个上下文对等待时间做出的判断，取消是任务生命周期收到的信号；前者允许在边界处转成业务结果，后者需要先清理并继续传播。

常见延伸问题

为什么在 timeout 内部捕获不到 TimeoutError？

因为超时上下文先在内部取消当前任务，再在退出上下文时完成异常转换。捕获点应放在 async with 外层。

主动取消一定不会变成 TimeoutError 吗？

不要依赖“异常名字”猜来源。由 timeout 自己触发并被其上下文识别的取消，才是超时转换路径；调用方主动取消应按 CancelledError 设计，并保留重新抛出的行为。

超时后还需要关闭连接吗？

需要。取消只改变控制流，不会替你完成业务对象的释放；资源拥有者仍应在 finally 中关闭响应、归还连接或撤销临时状态。