Python asyncio 怎样用绝对时间设置超时

在异步服务里，“最多等待 2 秒”和“必须在某个截止点前结束”不是同一个约束。asyncio.timeout(2) 接收相对时长，asyncio.timeout_at(deadline) 接收事件循环时钟上的绝对时间点。需要让一次请求的数据库、缓存和 HTTP 子操作共享总预算时，优先在入口计算一个 deadline，再向下传递。

官方地址：https://docs.python.org/3.14/library/asyncio-task.html

最稳妥的写法是：通过 asyncio.get_running_loop().time() 取得单调时钟，使用“当前时间 + 总预算”计算 deadline，再用 asyncio.timeout_at(deadline) 包住需要受约束的协程。TimeoutError 放在上下文管理器外捕获。

先分清 timeout 和 timeout_at 的输入含义

timeout(delay) 的参数是“从现在起再等多少秒”；timeout_at(when) 的参数则是“事件循环时钟走到哪个值时停止等待”。两者底层都属于 asyncio 的异步超时上下文管理器，但表达的约束层级不同。

这里的绝对时间不是 Unix 时间戳，也不应该用 datetime.now() 直接计算。asyncio 的 deadline 由事件循环的时钟解释，和 loop.time() 使用同一个时钟域，才能避免系统墙上时钟调整影响剩余时间。

用 loop.time() 计算一个可传递的 deadline

入口通常只做一次时间预算换算：读取当前事件循环时间，再加上本次请求允许的秒数。之后下层函数只接收 deadline，不再各自从当前时间重新加固定秒数。这样，调用链越深，剩余时间越少，约束仍然指向同一个截止点。

import asyncio


async def load_profile(deadline: float) -> dict:
    # 子函数接收同一时间域中的 deadline，不重复分配固定秒数。
    async with asyncio.timeout_at(deadline):
        await asyncio.sleep(0.05)
        return {"name": "示例用户"}


async def handle_request() -> dict | None:
    loop = asyncio.get_running_loop()
    # loop.time() 是事件循环的单调时钟；deadline 不是 Unix 时间戳。
    deadline = loop.time() + 2.0

    try:
        return await load_profile(deadline)
    except TimeoutError:
        # timeout_at 会在上下文退出时把取消转换成 TimeoutError。
        return None


asyncio.run(handle_request())

这个例子里，load_profile() 不需要知道入口给了多少秒，也不需要重新调用 loop.time()。它只关心传入的截止点。入口负责把超时转换成业务结果，子函数可以继续向更深层传递同一个值。

TimeoutError 要放在上下文管理器外面捕获

asyncio.timeout() 和 asyncio.timeout_at() 在超时时会取消当前任务，并在上下文管理器退出时把取消转换为内置的 TimeoutError。因此，把 try/except TimeoutError 写在 async with 外层更可靠；在上下文内部捕获这个异常，位置通常还没有完成转换。

import asyncio


async def fetch_slow_data() -> str:
    # 这里只模拟一个可能超过截止点的异步操作，不表示真实网络结果。
    await asyncio.sleep(3)
    return "data"


async def main() -> None:
    deadline = asyncio.get_running_loop().time() + 1.0
    try:
        async with asyncio.timeout_at(deadline):
            # 超时取消的是当前任务；资源清理仍应由被调用代码负责。
            value = await fetch_slow_data()
            print(value)
    except TimeoutError:
        # 异常在 async with 退出后处理，适合映射为超时响应或重试信号。
        print("请求超过统一截止时间")


asyncio.run(main())

不要把 TimeoutError 当成“后台任务一定停止”的证明。它表示当前超时上下文已经结束；被取消的协程仍应在自己的 finally 中关闭连接、释放锁或撤销临时资源。

把剩余预算传给嵌套的数据库和 HTTP 操作

固定给数据库 2 秒、再固定给 HTTP 2 秒，看似简单，却可能让一次请求总共等待 4 秒，超过入口原本允许的 2 秒。共享 deadline 的做法是让每一层都在同一时间点上工作，必要时把剩余时间换成局部的 timeout(remaining)。

import asyncio


async def query_database(remaining: float) -> list[dict]:
    # 局部调用只使用上层传来的剩余预算，避免重新开一份总预算。
    async with asyncio.timeout(remaining):
        await asyncio.sleep(0.02)
        return [{"id": 1}]


async def call_service(remaining: float) -> str:
    # 外部服务也受同一个请求预算约束，实际客户端还应处理连接清理。
    async with asyncio.timeout(remaining):
        await asyncio.sleep(0.02)
        return "ok"


async def compose_result() -> tuple[list[dict], str] | None:
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 1.0

    async def remaining_budget() -> float:
        # max 保证传入 timeout 的值不因时钟精度出现负数。
        return max(0.0, deadline - loop.time())

    try:
        rows = await query_database(await remaining_budget())
        state = await call_service(await remaining_budget())
        return rows, state
    except TimeoutError:
        # 任何一层耗尽同一 deadline，都统一从入口返回超时结果。
        return None


asyncio.run(compose_result())

如果两个子操作本来应该并发执行，可以把同一个 deadline 传给两个任务，再在外层统一处理异常；不要把上层 deadline 当成两个可以相加的独立预算。是否并发属于业务取舍，绝对截止点本身只表达时间边界。

不知道 deadline 时，用 timeout(None) 再 reschedule

有些流程在进入上下文时还拿不到上游的截止点，例如先读取一段配置，再根据配置计算剩余时间。此时可以用 asyncio.timeout(None) 创建一个暂不触发的上下文，拿到时间点后调用上下文对象的 reschedule()。

import asyncio


async def prepare_and_run() -> None:
    loop = asyncio.get_running_loop()
    try:
        # None 表示当前不设置截止点，后续可以重新安排。
        async with asyncio.timeout(None) as timeout_context:
            await asyncio.sleep(0)
            deadline = loop.time() + 1.0
            # reschedule 使用的仍然是 loop.time() 所在的时间域。
            timeout_context.reschedule(deadline)
            await asyncio.sleep(0.1)
    except TimeoutError:
        # 超时异常依然在上下文管理器外处理。
        print("准备阶段之后的操作超过 deadline")


asyncio.run(prepare_and_run())

Timeout 对象还可以通过 when() 查看当前 deadline，通过 expired() 判断上下文是否已经超过截止点。它们适合记录边界状态或做业务分支，但不应替代对 TimeoutError 的正常处理。

如果传入的 when 已经早于当前 loop.time()，官方语义是超时会在事件循环的下一次迭代触发。这个行为很适合表示“预算已经耗尽”，但调用方要预期到被包住的代码几乎不会获得额外执行时间。

什么时候改用 wait_for

asyncio.wait_for(awaitable, timeout) 适合单个 awaitable 的局部等待。它接收的是相对秒数；超时后会取消被等待对象，并抛出 TimeoutError。如果调用层已经有统一 deadline，先计算 remaining = max(0, deadline - loop.time())，再把剩余值交给 wait_for，比每一层都写一个固定秒数更一致。

import asyncio


async def one_operation() -> str:
    # 单个 awaitable 有独立封装需求时，wait_for 更直观。
    await asyncio.sleep(0.1)
    return "done"


async def call_one(deadline: float) -> str | None:
    loop = asyncio.get_running_loop()
    # 把绝对 deadline 换算成当前调用还能使用的相对预算。
    remaining = max(0.0, deadline - loop.time())
    try:
        return await asyncio.wait_for(one_operation(), timeout=remaining)
    except TimeoutError:
        # wait_for 的超时也是从调用处抛出，适合单点兜底。
        return None


async def main() -> None:
    loop = asyncio.get_running_loop()
    # deadline 必须在运行中的事件循环里计算，再传给单点调用。
    result = await call_one(loop.time() + 1.0)
    print(result)


asyncio.run(main())

在实际程序中不要手动创建或保存事件循环对象，应该在运行中的协程内用 get_running_loop() 计算 deadline，再调用 call_one()。若需要包住多个 await、嵌套上下文或统一处理取消，timeout_at() 通常更能表达设计意图。

常见误区与选择清单

• 不要把 datetime.now().timestamp() 当成 timeout_at() 的参数；这里要用 loop.time() 的时间域。
• 不要在每个下层函数里重新执行“当前时间 + 固定秒数”；这会让总耗时逐层累加。
• 不要在 async with 内部等待 TimeoutError；把异常处理放到上下文外。
• 需要动态获得时间点时，先用 timeout(None)，拿到 deadline 后再 reschedule()。
• 只包一个 awaitable 时可以选 wait_for()；要表达请求范围或嵌套关系时选 timeout_at()。

结语

绝对 deadline 的价值不在于语法更复杂，而在于把“请求必须在什么时候结束”变成可传递的约束。入口用 loop.time() + budget 建立时间点，下层复用它，异常在超时上下文外统一处理；这样数据库、缓存和服务调用不会各自偷偷扩张总预算。

相关问题

timeout_at 的 when 能不能传 Unix 时间戳？

不能直接这样做。when 应来自事件循环的时钟域，常见写法是 loop.time() + seconds。Unix 时间戳属于另一套墙上时钟。

timeout_at 超时后应该捕获 asyncio.TimeoutError 吗？

现代 Python 文档使用内置 TimeoutError。重点是把捕获位置放在 async with 外，而不是把异常处理写进超时上下文内部。

timeout 和 timeout_at 哪个更快？

两者不是性能选型题。按是否需要共享一次请求的绝对截止点选择：局部等待可用 timeout，共享总预算可用 timeout_at。