Python asyncio.timeout_at 如何用绝对截止时间包住任务

当一个异步请求要依次访问缓存、数据库和下游接口时，给每一段都写“再等 2 秒”很容易把总预算越用越长。asyncio.timeout_at() 的做法是先用事件循环的单调时钟算出一个绝对截止点，再让整个上下文共享它。这样无论里面有几个 await，任务都围绕同一个 deadline 收敛。

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

最小可靠写法是：用 loop.time() + 秒数 得到绝对时间，使用 async with asyncio.timeout_at(deadline) 包住工作，并在上下文外捕获 TimeoutError。不要把 time.time() 的 Unix 时间戳直接传给它。

先把相对超时换成事件循环时钟的绝对时间

timeout_at(when) 的 when 与 loop.time() 使用同一套单调时钟。单调时钟不受系统校时影响，适合计算持续时间。若当前预算是 3 秒，先保存 loop.time() + 3，后续所有异步操作都围绕这个值执行。

import asyncio

async def call_backend():
    # 绝对截止时间必须来自事件循环时钟，不能使用 time.time()。
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 3.0

    # 一个 deadline 覆盖同一项业务操作的全部 await。
    async with asyncio.timeout_at(deadline):
        return await asyncio.sleep(0.2, result="backend-ok")

asyncio.run(call_backend())

这里的关键不是把 3 秒写进 API，而是把业务预算转换成事件循环时钟上的一个点。若把 time.time() 传入，数值量级通常完全不同，可能一进入上下文就被判断为已过期。

把多个 await 放在同一个 timeout_at 边界内

绝对截止时间特别适合“先做 A，再做 B，但总耗时不能超过 X 秒”的场景。上下文内部可以有多个 await，第二个操作不会重新获得一份完整预算。

async def load_page():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 1.5

    # 两个阶段共用总预算，避免每个阶段各自重新计时。
    async with asyncio.timeout_at(deadline):
        profile = await read_profile()
        orders = await read_orders(profile["user_id"])
        return {"profile": profile, "orders": orders}

如果截止时间来自上层请求，就把同一个 deadline 作为参数向下传递，而不是在每个函数里重新调用 loop.time() + timeout。这样更容易做链路级超时控制，也能避免嵌套调用悄悄延长总时限。

超时异常为什么要放在 async with 外面

超时管理器会取消当前任务，内部看到的是 asyncio.CancelledError；离开上下文时，管理器再把这次取消转换为 TimeoutError。因此下面的 except TimeoutError 必须包住整个 async with，写在上下文内部通常捕获不到这个转换后的异常。

async def fetch_with_fallback():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 0.8

    try:
        async with asyncio.timeout_at(deadline):
            return await slow_service()
    except TimeoutError:
        # 这里只处理当前超时；真实项目可返回降级数据或记录指标。
        return {"source": "fallback", "items": []}

不要在协程内部无条件吞掉 CancelledError。上层取消、任务组退出和超时都可能经过取消机制；如果业务确实需要清理资源，应在 finally 中清理，然后让取消继续传播。

用 expired() 判断结果，而不是只看返回值

把上下文对象接出来，就能在收尾阶段调用 expired()。它回答的是“这个超时上下文是否越过了 deadline”，适合写日志、指标或区分“业务返回为空”和“时间预算耗尽”。

async def run_job():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 2.0
    cm = None

    try:
        async with asyncio.timeout_at(deadline) as current_timeout:
            cm = current_timeout
            await do_job()
    except TimeoutError:
        # 异常在上下文外处理，避免混淆取消和超时转换。
        pass
    finally:
        if cm is not None and cm.expired():
            record_timeout_metric()

expired() 是状态判断，不会把已经成功返回的任务重新标记为超时。嵌套的 timeout 上下文也是安全的，但工程上最好让每一层的责任清楚：外层控制请求总预算，内层只负责更窄的依赖预算。

timeout_at、timeout 和 wait_for 怎么选

如果上层已经计算出请求截止点，优先传递 deadline；如果只有“这一段最多等 500 毫秒”，timeout(0.5) 更直观；只包一个 awaitable 且不需要上下文组合时，再考虑 wait_for。这些 API 自 Python 3.11 起的超时上下文能力才可用，部署到旧版本前要先确认解释器环境。

常见问题

asyncio.timeout_at 可以传 time.time() 吗？

不建议。它要求事件循环时钟上的绝对值，应使用 asyncio.get_running_loop().time() 计算 deadline；time.time() 是 Unix 墙上时钟，两者不是同一基准。

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

因为上下文内部先处理取消，离开上下文时才完成到 TimeoutError 的转换。把 try/except 放在包住 async with 的外层即可。

expired() 返回 True 就代表业务一定失败吗？

它只说明这个超时上下文越过了截止点。业务是否失败仍取决于调用方如何降级、重试或返回结果，不能把状态判断当成业务结论。