Python asyncio.timeout_at怎样按绝对时间设置异步截止点

如果多个异步步骤需要共享同一个“最晚完成时间”，asyncio.timeout_at() 比分别给每一步传入相对秒数更直观。它接收的是事件循环时钟上的绝对截止点，不是 Unix 时间戳；常见写法是先取 loop.time()，再加上允许等待的秒数：

import asyncio


async def load_profile():
    # 这是示例协程，模拟一个可能较慢的异步操作。
    await asyncio.sleep(1)
    return {"name": "demo"}


async def main():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 5

    try:
        # timeout_at 使用事件循环时钟上的绝对截止点。
        async with asyncio.timeout_at(deadline):
            profile = await load_profile()
            print(profile)
    except TimeoutError:
        # 超时异常应在 async with 外部捕获。
        print("读取资料超时")


asyncio.run(main())

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

timeout_at(deadline) 的关键不是“等待 5 秒”这个数字，而是让后续多个 await 共同面对同一个 deadline。这样前一个步骤已经消耗的时间不会被后一个步骤重新计算一遍。

• deadline 用 loop.time() 计算，不能把 time.time() 的 Unix 时间直接传进去。
• TimeoutError 在 async with 外捕获，内部的取消转换由超时上下文管理器负责。
• 已有相对秒数时可用 asyncio.timeout()；已有绝对 deadline 时用 asyncio.timeout_at()。

用 loop.time() 计算绝对截止点

asyncio.timeout_at(when) 的 when 是事件循环时钟上的绝对时间。它与墙上时钟的 Unix 时间戳不是同一套坐标，因此不要把 time.time() + 5 直接交给它。需要“从现在起允许 5 秒”时，应在同一个事件循环上完成换算：

import asyncio


async def fetch_piece(name):
    # 模拟每个子步骤都需要等待的异步工作。
    await asyncio.sleep(0.8)
    return name


async def main():
    loop = asyncio.get_running_loop()
    # 相对时长先转换成同一事件循环时钟上的绝对 deadline。
    deadline = loop.time() + 5

    try:
        # 两个 await 共用一个总截止点，而不是各自重新拥有 5 秒。
        async with asyncio.timeout_at(deadline):
            first = await fetch_piece("header")
            second = await fetch_piece("body")
            print(first, second)
    except TimeoutError:
        # 任一步骤让总 deadline 过期，统一进入超时分支。
        print("总截止点已到")


asyncio.run(main())

这里的 deadline 只在当前事件循环的时间坐标内有意义。把它保存到数据库、跨进程传输，或拿到另一个事件循环里继续使用，都不能保证语义仍然成立。若业务协议需要传递公共截止时刻，应先定义协议使用的时钟，再在每个事件循环内换算成自己的 deadline。

在 async with 外处理 TimeoutError

超时上下文管理器会取消当前任务的等待，并在上下文管理器内部处理取消过程，最终把超时表现为 TimeoutError。因此，直接在 async with 的代码块里捕获它，通常不是正确的结构；应该让上下文退出，再在外层统一捕获：

import asyncio


async def slow_operation():
    # 让示例操作明显超过调用方给出的 deadline。
    await asyncio.sleep(10)


async def run_with_deadline():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 0.2

    try:
        # 超时发生后，退出这个上下文再处理 TimeoutError。
        async with asyncio.timeout_at(deadline):
            await slow_operation()
    except TimeoutError:
        # 这里适合记录超时、返回降级结果或通知上层重试。
        return "fallback"

    return "ok"


async def main():
    result = await run_with_deadline()
    print(result)


asyncio.run(main())

这段结构表达了一个重要边界：TimeoutError 是超时上下文对取消结果的外部表现，不要把它和协程内部主动抛出的同名异常混为一谈。若需要区分业务异常和超时，建议让业务异常在上下文外继续按原有层级处理，并给超时分支单独的降级策略。

区分 timeout、timeout_at、None 与过期 deadline

两个 API 的区别可以压缩成一个判断：如果手里是“还能等多少秒”，用 asyncio.timeout(delay)；如果手里已经有事件循环时钟上的“最晚时间点”，用 asyncio.timeout_at(when)。两者都返回异步上下文管理器，都可以嵌套使用。

None 不是“立即超时”，而是表示当前没有截止点；如果传入的绝对时间已经早于 loop.time()，则上下文会在事件循环下一次迭代中触发。这样可以让调用方在已有 deadline 过期时仍走统一的超时处理路径。

用 Timeout 对象调整和检查截止点

有些请求在开始时还不知道最终时限，例如要先读取配置或等待上游返回配额。这时可以先创建一个没有 deadline 的上下文，拿到 Timeout 对象后再调用 reschedule()；退出上下文后还可以用 expired() 查询是否越过了截止点：

import asyncio


async def read_config():
    # 示例中用短暂等待模拟异步读取配置。
    await asyncio.sleep(0.05)
    return 1.0


async def main():
    loop = asyncio.get_running_loop()
    timeout_state = None

    try:
        # 尚不知道最终时限时先创建一个不触发的 Timeout。
        async with asyncio.timeout(None) as timeout_state:
            seconds = await read_config()
            # reschedule 接收的是绝对 deadline，而不是相对秒数。
            timeout_state.reschedule(loop.time() + seconds)
            await asyncio.sleep(0.2)
    except TimeoutError:
        # 超时仍然应在上下文外处理。
        print("动态 deadline 已过期")
    finally:
        # 对象存在时可以判断这次上下文是否越过了 deadline。
        if timeout_state is not None:
            print("expired:", timeout_state.expired())


asyncio.run(main())

reschedule() 和 timeout_at() 使用相同的绝对时间语义。不要把 seconds 本身传给 reschedule()；应先加上当前 loop.time()。如果 deadline 已经过去，重排后的上下文会在后续事件循环迭代中进入超时路径。

常见问题与使用结论

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

不建议。timeout_at 的时间基准是当前事件循环的 loop.time()，而 time.time() 返回 Unix 时间戳；两者零点和用途不同。应使用 loop.time() + 相对秒数 计算 deadline。

为什么 TimeoutError 要写在 async with 外面？

超时上下文在内部处理任务取消，并在退出上下文时把结果转换成 TimeoutError。把捕获放在外面，才能接住这个由上下文管理器转换后的异常。

timeout_at 的 deadline 已经过期会怎样？

如果 when 早于当前事件循环时间，它会在下一次事件循环迭代触发，而不是依赖一个负数的相对等待时长。

什么时候应该用 timeout 而不是 timeout_at？

单个局部操作只需要“从现在起最多等待几秒”时，timeout 更易读；多个步骤需要共享上游传来的最晚时刻时，timeout_at 更能避免每层重复计算预算。

实际使用时可以按这个顺序决定：先确认时间基准是事件循环时钟，再判断手里是相对时长还是绝对 deadline；把 TimeoutError 的处理放在上下文外，需要动态时限就用 None、reschedule() 和 expired() 组合。这样既能让多个 await 共享一个总预算，也能避免把墙上时间和事件循环时间混用。