Python asyncio.timeout 怎么动态调整截止时间

可以动态调整。关键是把 asyncio.timeout() 返回的上下文管理器保存为变量，然后调用 reschedule(when)。其中 when 不是“再等多少秒”，而是事件循环时钟上的绝对截止时间，所以常用写法是 asyncio.get_running_loop().time() + 秒数。

import asyncio

async def run_with_dynamic_timeout(work, budget_seconds):
    loop = asyncio.get_running_loop()

    try:
        async with asyncio.timeout(None) as timeout:
            # 预算确定后，把相对秒数换成事件循环的绝对截止时间
            timeout.reschedule(loop.time() + budget_seconds)
            return await work()
    except TimeoutError:
        # TimeoutError 要在 async with 外捕获
        return None

如果一开始就知道预算，也可以先写 asyncio.timeout(10)，之后仍然通过 reschedule() 改成新的截止时间。官方文档地址：https://docs.python.org/3/library/asyncio-task.html。该组 API 从 Python 3.11 开始提供。

为什么动态场景常从 timeout(None) 开始

很多异步任务在进入处理逻辑时还不知道最终预算。例如，请求头里携带剩余时间，或者上游元数据返回后才能决定当前操作最多运行多久。这时用 asyncio.timeout(None) 创建上下文不会立即设置期限，但会返回一个可操作的 Timeout 对象；拿到预算后，再给它安排截止时间。

下面的例子模拟“先读取任务元数据，再决定剩余预算”。元数据读取阶段也位于同一个上下文中，但在调用 reschedule 之前没有超时限制：

import asyncio

async def load_policy():
    # 示例：异步读取上游给出的预算
    await asyncio.sleep(0.05)
    return {"budget_seconds": 1.5}

async def fetch_data():
    # 示例：真正需要受截止时间约束的工作
    await asyncio.sleep(0.2)
    return {"ok": True}

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

    try:
        async with asyncio.timeout(None) as timeout:
            policy = await load_policy()
            seconds = policy["budget_seconds"]

            # reschedule 需要绝对时刻，不是直接传 1.5
            timeout.reschedule(loop.time() + seconds)
            result = await fetch_data()
            print(result)
    except TimeoutError:
        # 上下文退出时才会把内部取消转换成 TimeoutError
        print("任务超过动态截止时间")

    if timeout is not None:
        # expired 可用于确认这个超时上下文是否已经过期
        print("expired:", timeout.expired())

asyncio.run(main())

如何缩短、延长或取消截止时间

reschedule() 每次都会替换当前截止时间。因此，传入更近的绝对时刻就是缩短，传入更远的绝对时刻就是延长，传入 None 则表示暂时不触发超时。可以用 when() 读取当前截止时间。

import asyncio

async def adjustable_job():
    loop = asyncio.get_running_loop()

    async with asyncio.timeout(10) as timeout:
        # 缩短：从当前时刻起只再给 2 秒
        timeout.reschedule(loop.time() + 2)
        print("short deadline:", timeout.when())

        # 延长：重新设置为从当前时刻起 8 秒
        timeout.reschedule(loop.time() + 8)
        print("extended deadline:", timeout.when())

        # 取消当前期限：后续不会由这个上下文触发超时
        timeout.reschedule(None)
        print("disabled:", timeout.when())

        await asyncio.sleep(0.1)

如果传入的绝对时刻早于 loop.time()，超时会在事件循环的下一次迭代触发。这个行为适合表达“预算已经耗尽，尽快结束”，但它并不是同步地在 reschedule() 这一行立刻抛出异常。

最常见的错误：把相对秒数直接传给 reschedule

下面的写法看似表示“5 秒后超时”，实际上传入的是事件循环时钟上的绝对值 5。正常运行中的 loop.time() 通常早已大于 5，所以它会被视为已经过期，并在下一轮事件循环触发超时。

async def wrong_and_right():
    loop = asyncio.get_running_loop()

    async with asyncio.timeout(None) as timeout:
        # 错误：5 会被当作绝对时刻，不是 5 秒时长
        # timeout.reschedule(5)

        # 正确：先取事件循环当前时刻，再加 5 秒
        timeout.reschedule(loop.time() + 5)
        await asyncio.sleep(0.1)

不要用 time.time() 拼接截止时间。Timeout 明确使用事件循环自己的单调时钟，统一使用 loop.time() 可以避免系统时间校准或手工修改带来的跳变。

为什么 TimeoutError 必须在 async with 外捕获

截止时间到达时，超时上下文会取消当前任务，并在内部接收由此产生的 CancelledError；退出上下文时，它才把这次取消转换为内置的 TimeoutError。因此，except TimeoutError 应包住整个 async with，而不是写在上下文内部。

import asyncio

async def call_service():
    await asyncio.sleep(2)

async def guarded_call():
    timeout = None

    try:
        async with asyncio.timeout(None) as timeout:
            loop = asyncio.get_running_loop()
            # 只给服务调用 0.2 秒
            timeout.reschedule(loop.time() + 0.2)
            await call_service()
    except TimeoutError:
        # 此时已经离开 timeout 上下文，可以检查它是否过期
        if timeout is not None and timeout.expired():
            return "deadline exceeded"
        raise

    return "ok"

这里不要在协程内部随意吞掉 CancelledError。如果业务代码截获取消却不继续抛出，超时上下文可能无法按预期完成取消语义，也会让结构化并发更难判断任务状态。

固定绝对截止时间可直接用 timeout_at

若多个阶段需要共享同一个绝对期限，可以先计算一次 deadline，再使用 asyncio.timeout_at(deadline)。它与 asyncio.timeout() 使用相同的事件循环时钟，区别只是入口直接接收绝对时刻。

import asyncio

async def pipeline():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 3

    try:
        async with asyncio.timeout_at(deadline):
            # 两个阶段共享同一个总截止时间
            await asyncio.sleep(0.1)
            await asyncio.sleep(0.2)
    except TimeoutError:
        return "pipeline timeout"

    return "done"

超时上下文可以安全嵌套，所以外层可以限制整条请求链，内层限制某个数据库或网络步骤。内层期限应根据当前剩余预算计算，避免把局部操作设置得比外层总期限更宽松。

相关问题

reschedule 可以调用多次吗？ 可以。每次调用都会替换当前截止时间，但应在超时上下文仍处于活动状态时调整。

when() 返回什么？ 返回事件循环时钟上的当前绝对截止时间；未设置期限时返回 None。

expired() 能代替捕获 TimeoutError 吗？ 不能。真正到期时仍会在上下文外抛出 TimeoutError；expired() 适合在异常处理后确认该上下文是否已经过期。

动态调整后会取消哪个任务？ asyncio.timeout() 限制的是当前任务在上下文中的等待；到期时先取消当前任务，再由上下文把相应取消转换为 TimeoutError。