Python asyncio.timeout 怎样把嵌套任务限制在总时限内

要让嵌套异步调用共享总时限，把整个请求的等待范围放进外层 asyncio.timeout(total)，局部操作再设置较短的 timeout；并发子任务由范围内的 TaskGroup 管理。内层超时只处理允许降级的局部失败，外层到期则中断整个请求。不要在每个串行阶段重新给一份完整总预算。

官方地址：https://docs.python.org/3/

特性解决什么：总预算不是各层等待时间相加

一个响应可能既要加载必需数据，也要加载允许缓存降级的辅助数据。总预算约束的是这次响应从进入范围到完成的等待；局部预算则限制某个依赖占用多久。内层写了 5 秒，不代表外层只剩 1 秒时它还能继续获得 5 秒。

反过来，如果三个串行阶段各自建立一个新的 2 秒 timeout，却没有外层约束，那么它们可以分别耗尽各自的预算。正确做法是让一个外层作用域包含所有属于请求的阶段，或者在跨函数边界时传递同一个绝对截止时间。

支持范围：timeout 管当前任务，TaskGroup 管子任务

asyncio.timeout、timeout_at 和 TaskGroup 从 Python 3.11 提供。本文使用这些公共 API，事实说明参考 Python 3.14 官方文档，没有依赖更高版本新增的任务组接口。低版本项目应明确升级计划或另行设计取消与收尾，不能直接复制这段代码。

timeout 到期时取消的是当前任务；退出上下文时，它把由自身超时导致的取消转换为内置 TimeoutError。因此应在对应的 async with 外捕获 TimeoutError。子任务范围则交给 TaskGroup：退出组时等待任务收尾，避免只返回一个超时状态，却把仍在运行的协程遗留在后台。

最小示例：局部降级不放大请求预算

示例里必需数据模拟等待 0.25 秒，可选数据模拟等待 0.50 秒，但可选数据的局部时限只有 0.15 秒。第一个请求总预算 1 秒，第二个只有 0.10 秒。这些等待值用于展示分支关系，不是实际网络延迟或性能测量。

import asyncio

async def load_required():
    # 用可取消的异步等待模拟必需的数据源。
    await asyncio.sleep(0.25)
    return {"name": "demo"}

async def load_optional():
    local = asyncio.timeout(0.15)
    try:
        async with local:
            await asyncio.sleep(0.50)
            return ["remote"]
    except TimeoutError:
        # 只把本层到期转为降级，不掩盖下游自己的超时异常。
        if not local.expired():
            raise
        return ["cached"]

async def build_response(total_seconds):
    total = asyncio.timeout(total_seconds)
    try:
        async with total:
            # 子任务归属同一组，总时限覆盖启动、等待和结果组装。
            async with asyncio.TaskGroup() as group:
                required = group.create_task(load_required())
                optional = group.create_task(load_optional())
            return {
                "status": "ok",
                "required": required.result(),
                "optional": optional.result(),
            }
    except TimeoutError:
        if not total.expired():
            raise
        # 只在整个请求确实到期时返回超时结果。
        return {"status": "timeout"}

async def main():
    # 参数仅用于对比局部降级与总预算到期，不是性能基准。
    print(await build_response(1.0))
    print(await build_response(0.10))

asyncio.run(main())

按规则推导，在事件循环能正常调度、没有其他异常的情况下，第一个调用可以得到必需数据，并将可选部分降级为 cached；第二个调用走总超时结果。实际调度与清理会影响观察到的耗时，不能把这些小数当作精确返回时间承诺。

注意每个 TimeoutError 分支还检查了自己的 expired()。因为底层依赖也可能主动抛出 TimeoutError，不能见到同名异常就断言本层计时器到期。这里让其他来源的超时继续传播，也没有用一个空列表掩盖所有异常。

兼容与结果处理：先区分谁到期，再决定是否保留结果

局部 timeout 的取消只在它自己的范围退出时转换为 TimeoutError，因此 load_optional 能处理局部到期并返回缓存值。若总预算先到期，子任务收到的取消不能在内部被吞掉；它需要继续传播，让任务组完成收尾。不要在辅助函数里写捕获 BaseException 后直接返回成功值的兜底。

任务组正常退出后再读取 result()。总超时分支不会把已经完成的一部分任务拼成完整成功响应；如果业务允许部分结果，应明确每个部分的状态和可缺省字段，并在设计里区分“已完成”“失败”“被取消”，而不是通过访问未完成任务的 result 猜测。

子任务出现非取消异常时，TaskGroup 会取消其他任务，并可能以 ExceptionGroup 汇总失败。外层普通 except TimeoutError 不会自动处理其中的 TimeoutError；需要按实际业务在任务内部转换明确的可降级失败，或由上层使用合适的异常组处理。清理期间出现其他异常时，也不能坚持把结果一律标为总超时。

对于多层串行调用，可以只在入口计算一次 deadline = asyncio.get_running_loop().time() + budget，下层通过 asyncio.timeout_at(deadline) 使用相同截止点。局部限制可取 min(deadline, loop.time() + local_limit)。这个绝对值必须来自同一事件循环的时钟域，不能把 time.time 的日历时间混进去，也不应序列化后跨进程直接复用。

性能注意：总时限是协作取消边界

CPU 密集循环或同步阻塞调用占住事件循环时，超时回调也无法及时执行。即便代码写了 1 秒，阻塞函数仍可能让整个请求延迟更久。应让业务在合适位置进行异步等待，或按工作性质采用合适执行方式；将操作放进线程也不意味着取消等待就能强制终止线程里的工作。

取消需要给 finally 中的资源清理留出时间，TaskGroup 也要等其子任务退出。因此 deadline 表示开始触发协作取消的边界，不能当作进程级硬停止开关。清理逻辑要短小、有资源所有权，避免在 finally 中启动无期限重试。

asyncio.shield 会保护被等待的任务不随调用方取消，和“请求结束时全部停止”的目标可能冲突。确实需要独立后台操作时，应由另外的生命周期管理者持有任务引用、负责异常与退出；不能仅为了绕过 timeout 就随手 shield。

相关问题

内层捕获 TimeoutError 会吃掉外层取消吗

正常嵌套时，外层取消不会变成内层计时器自己的到期。不要额外吞掉 CancelledError；如果依赖抛出同名异常，还应结合 expired 判断来源。

直接 create_task 再离开 timeout，子任务一定停止吗

不能只凭创建位置判断。timeout 管当前任务，而任务是否随取消传播取决于等待关系与管理方式。本文把并发工作交给范围内的 TaskGroup，明确负责收尾。

总预算很短，是否应该重试直到成功

重试也必须消耗同一预算。不要在超时捕获后跳出总范围重新启动完整重试窗口；是否重试、是否幂等及剩余时间是否足够，应由请求策略决定。

参考资料

• Python：任务组、超时、取消与 shield
• Python：CancelledError 与 TimeoutError
• Python：事件循环与时钟