usehook
创建时间: 2026-08-25最后更新: 2026-08-25

1. 定时

异步程序经常需要暂时等待：一秒后重试请求、三十秒后判定超时、每隔一段时间刷新缓存，或者在指定时间执行一次回调。这些需求看起来都是「过一段时间再运行」，但实际包含三类不同问题：

需求常用方式
当前协程暂停一段时间await asyncio.sleep()
将普通函数安排到未来执行loop.call_later()、loop.call_at()
限制异步操作的最长等待时间asyncio.timeout()、asyncio.wait_for()

asyncio 不会为每个定时器创建一个线程。事件循环会记录定时器的截止时间，计算自己最多能等待多久，并在时间到达后把对应的 Task 或回调放入可运行队列。

定时器到期只表示工作已经具备运行条件，不表示代码一定会在那个时刻精确执行。如果事件循环当时正在运行耗时的同步代码，或者还有其他就绪任务需要处理，这项工作只能等事件循环再次取得控制权。

timer-delay.py
01
import asyncio

02
import time

03


04
async def main():

05
    started_at = time.perf_counter()

06
    await asyncio.sleep(1)

07


08
    elapsed = time.perf_counter() - started_at

09
    print(f"实际等待：{elapsed:.3f} 秒")

10


11
asyncio.run(main())

输出通常在一秒附近，并可能因为调度开销略大于一秒。这里的一秒应该理解为目标延迟，而不是实时系统提供的精确执行承诺。

2. sleep

asyncio.sleep(delay) 是应用代码中最常用的延迟方式。它会暂停当前 Task，并允许事件循环继续推进其他 Task：

sleep.py
01
import asyncio

02


03
async def remind(name, delay):

04
    print(f"{name} 开始等待")

05
    await asyncio.sleep(delay)

06
    print(f"{name} 继续执行")

07


08
async def main():

09
    await asyncio.gather(

10
        remind("任务 A", 1),

11
        remind("任务 B", 0.2),

12
    )

13


14
asyncio.run(main())

两个任务会先后开始等待。任务 B 的时间先到，因此通常先恢复；任务 A 在这段时间里并没有占用一个线程空等。

sleep() 接收的延迟单位是秒，可以使用整数或小数。它还可以接收一个 result 参数，并在等待结束后返回这个值：

sleep-result.py
1
import asyncio

2


3
async def main():

4
    result = await asyncio.sleep(0.1, result="继续执行")

5
    print(result)

6


7
asyncio.run(main())

这种返回值用法并不常见，大多数代码只把 sleep() 当作等待点。

主动让出执行权

await asyncio.sleep(0) 不会进行有意义的时间延迟，但仍然会暂停当前 Task，让事件循环有机会推进其他已就绪任务：

sleep-zero.py
01
import asyncio

02


03
async def calculate():

04
    total = 0

05


06
    for number in range(1_000_000):

07
        total += number

08


09
        if number % 100_000 == 0:

10
            await asyncio.sleep(0)

11


12
    return total

13


14
asyncio.run(calculate())

这可以避免一段较长的 Python 循环始终占住事件循环，但它不会让 CPU 计算变快，也不会把计算变成并行任务。CPU 密集型工作通常应该拆分、优化，或者交给其他进程，而不是不断插入 sleep(0)。

阻塞等待

time.sleep() 是同步阻塞函数。在事件循环线程中调用它时，不只是当前协程暂停，整个事件循环都无法调度其他 Task：

blocking-sleep.py
01
import asyncio

02
import time

03


04
async def bad_task(name):

05
    print(f"{name} 开始")

06
    time.sleep(1)

07
    print(f"{name} 完成")

08


09
async def main():

10
    started_at = time.perf_counter()

11


12
    await asyncio.gather(

13
        bad_task("任务 A"),

14
        bad_task("任务 B"),

15
    )

16


17
    print(f"总耗时：{time.perf_counter() - started_at:.1f} 秒")

18


19
asyncio.run(main())

这两个任务的总耗时接近两秒，因为任务 A 调用 time.sleep() 后，任务 B 连开始执行的机会都没有。把它改成 await asyncio.sleep(1) 后，两个 Task 可以在同一段时间内等待，总耗时会接近一秒。

同样需要注意：没有 await 的长循环、大量 JSON 解析或其他耗时同步函数，也会像 time.sleep() 一样拖延所有定时器。

3. 回调

如果需要在未来执行一个普通函数，可以使用事件循环的 call_later()：

call-later.py
01
import asyncio

02


03
def show_message(message):

04
    print(message)

05


06
async def main():

07
    loop = asyncio.get_running_loop()

08
    loop.call_later(1.5, show_message, "一段时间后执行")

09
    print("回调已经注册")

10


11
    await asyncio.sleep(2)

12


13
asyncio.run(main())

call_later(delay, callback, *args) 的参数可以这样理解：

参数含义
delay从现在开始等待多少秒
callback到期后调用的普通函数
*args调用回调时传入的位置参数

回调只会被调用一次。如果多个回调被安排在完全相同的时间点，官方并不保证它们之间的执行顺序，因此业务逻辑不应依赖这个顺序。

call_later() 返回的是 TimerHandle，不是 Task。它不能被 await，也不会替调用者保存回调的返回值。回调发生异常时，异常会交给事件循环的异常处理机制，而不是像 await task 那样直接抛回当前协程。因此，需要结果、异常传播和取消链路时，通常应该使用协程与 Task。

这些底层调度方法不能直接传递关键字参数。需要关键字参数时，可以使用 functools.partial()：

callback-keywords.py
01
import asyncio

02
from functools import partial

03


04
def show_message(message, *, prefix):

05
    print(f"{prefix} {message}")

06


07
async def main():

08
    loop = asyncio.get_running_loop()

09
    callback = partial(show_message, "缓存已刷新", prefix="[定时器]")

10
    loop.call_later(0.5, callback)

11


12
    await asyncio.sleep(1)

13


14
asyncio.run(main())

相比匿名函数，partial() 通常能在调试信息中提供更清楚的回调描述。

TimerHandle

call_later() 返回一个 TimerHandle，它代表刚刚注册的定时回调：

timer-handle.py
01
import asyncio

02


03
def show_message():

04
    print("这条消息不会出现")

05


06
async def main():

07
    loop = asyncio.get_running_loop()

08
    handle = loop.call_later(2, show_message)

09


10
    print(handle.when() - loop.time())

11


12
    handle.cancel()

13
    print(handle.cancelled())  # True

14


15
    await asyncio.sleep(2.2)

16


17
asyncio.run(main())

cancel() 可以取消尚未执行的回调，重复取消不会产生额外影响；cancelled() 用于检查回调是否被取消；when() 返回回调原本计划执行的事件循环时间。

如果回调已经执行，再调用 cancel() 无法撤销已经发生的操作。取消 TimerHandle 也只会取消这一次回调，不会自动取消回调中另外创建的 Task。

4. 时间

call_later() 使用相对延迟，而 call_at() 使用事件循环时钟上的绝对时间点：

call-at.py
01
import asyncio

02


03
def show_message():

04
    print("到达计划时间")

05


06
async def main():

07
    loop = asyncio.get_running_loop()

08
    deadline = loop.time() + 1

09


10
    handle = loop.call_at(deadline, show_message)

11
    print(handle.when() == deadline)  # True

12


13
    await asyncio.sleep(1.2)

14


15
asyncio.run(main())

loop.time() 返回事件循环内部使用的单调时钟。它适合计算「过去了多久」或「还剩多久」，但这个数字不是 Unix 时间戳，也不能转换成日历日期。

如果传给 call_at() 的时间已经过去，回调会在事件循环下一次有机会调度时尽快执行。出于性能考虑，定时回调理论上可能在单调时钟一个分辨率范围内略早触发；事件循环繁忙时则可能明显延后。因此，不要用它实现依赖毫秒级精度的实时控制。

单调时钟的重要特点是不会因为用户修改系统时间、网络校时或夏令时变化而突然跳跃。因此，计算延迟和截止时间时，应使用同一种时钟：

monotonic-time.py
01
import asyncio

02


03
async def main():

04
    loop = asyncio.get_running_loop()

05
    started_at = loop.time()

06


07
    await asyncio.sleep(0.5)

08


09
    elapsed = loop.time() - started_at

10
    print(f"经过了 {elapsed:.1f} 秒")

11


12
asyncio.run(main())

不要把 time.time()、datetime.now().timestamp() 得到的墙上时钟时间直接传给 call_at()。它们和 loop.time() 不使用同一个时间基准。

如果需求是「今天 18:00 执行」，可以先用 datetime 计算它距离现在还有多少秒，再使用 call_later()：

wall-clock-delay.py
01
import asyncio

02
from datetime import datetime, timedelta

03


04
def show_message():

05
    print("到达目标时间")

06


07
async def main():

08
    loop = asyncio.get_running_loop()

09


10
    target = datetime.now() + timedelta(seconds=1)

11
    delay = max(0, (target - datetime.now()).total_seconds())

12


13
    loop.call_later(delay, show_message)

14
    await asyncio.sleep(delay + 0.2)

15


16
asyncio.run(main())

这种方式适合进程存活期间的一次短期安排。需要跨进程重启保存、按时区运行、错过后补偿或在多台服务器中只执行一次的日历任务，应使用持久化调度器或任务队列。

5. 协程

call_later() 和 call_at() 的 callback 必须是普通可调用对象。直接传入异步函数并不会自动等待它：

wrong-async-callback.py
01
import asyncio

02


03
async def send_reminder():

04
    print("发送提醒")

05


06
async def main():

07
    loop = asyncio.get_running_loop()

08


09
    # 错误：到期后只会创建协程对象，没有代码负责 await 它

10
    # loop.call_later(1, send_reminder)

11


12
    await asyncio.sleep(1.2)

13


14
asyncio.run(main())

异步函数被普通回调调用时只会返回协程对象，这个对象不会自行运行，还可能产生 coroutine was never awaited 警告。

应用代码需要延迟执行协程时，更自然的方式是创建一个负责等待的协程，再把它包装成 Task：

delayed-coroutine.py
01
import asyncio

02


03
async def send_reminder(message):

04
    await asyncio.sleep(0.1)

05
    print(message)

06


07
async def run_later(delay, async_function, *args):

08
    await asyncio.sleep(delay)

09
    return await async_function(*args)

10


11
async def main():

12
    task = asyncio.create_task(

13
        run_later(1, send_reminder, "该休息一下了"),

14
        name="delayed-reminder",

15
    )

16


17
    print("提醒任务已经创建")

18
    await task

19


20
asyncio.run(main())

这样得到的是正常 Task，可以等待结果、接收异常，也可以在提醒尚未执行时取消整条调用链。

底层库确实需要从定时回调启动协程时，可以让普通回调调用 create_task()，但仍然要保存 Task 引用并处理异常。普通业务代码优先使用上面的 sleep() 加 Task 结构，会更容易管理生命周期。

6. 超时

延迟解决的是「多久以后继续」，超时解决的是「最多允许等待多久」。从 Python 3.11 开始，可以使用 asyncio.timeout() 给一段异步代码设置时间限制：

timeout.py
01
import asyncio

02


03
async def fetch():

04
    await asyncio.sleep(2)

05
    return "数据"

06


07
async def main():

08
    try:

09
        async with asyncio.timeout(1):

10
            result = await fetch()

11
            print(result)

12
    except TimeoutError:

13
        print("请求超时")

14


15
asyncio.run(main())

超时发生时，asyncio.timeout() 会取消当前 Task，在上下文管理器内部处理 CancelledError，并把它转换成 TimeoutError。因此，TimeoutError 要在 async with 外部捕获。

一个超时范围可以包含多个异步操作，限制它们共同消耗的总时间：

timeout-block.py
01
import asyncio

02


03
async def step(name, delay):

04
    await asyncio.sleep(delay)

05
    print(f"{name} 完成")

06


07
async def main():

08
    try:

09
        async with asyncio.timeout(1):

10
            await step("第一步", 0.6)

11
            await step("第二步", 0.6)

12
    except TimeoutError:

13
        print("整个流程超时")

14


15
asyncio.run(main())

每一步单独看都没有超过一秒，但两步合计超过一秒，因此整个代码块会超时。这很适合限制一次接口请求或一条 Agent 执行链的总耗时。

如果已经计算出事件循环时钟上的截止时间，可以使用 asyncio.timeout_at()：

timeout-at.py
01
import asyncio

02


03
async def main():

04
    loop = asyncio.get_running_loop()

05
    deadline = loop.time() + 1

06


07
    try:

08
        async with asyncio.timeout_at(deadline):

09
            await asyncio.sleep(2)

10
    except TimeoutError:

11
        print("超过截止时间")

12


13
asyncio.run(main())

只想限制一个可等待对象时，也可以使用 asyncio.wait_for()：

wait-for.py
01
import asyncio

02


03
async def fetch():

04
    await asyncio.sleep(2)

05
    return "数据"

06


07
async def main():

08
    try:

09
        result = await asyncio.wait_for(fetch(), timeout=1)

10
        print(result)

11
    except TimeoutError:

12
        print("请求超时")

13


14
asyncio.run(main())
方式适合场景
asyncio.timeout()限制一个异步代码块的总时间
asyncio.timeout_at()已经有基于 loop.time() 的绝对截止时间
asyncio.wait_for()限制单个可等待对象

wait_for() 超时后会取消被等待对象，并等待它完成取消流程，所以函数实际返回的时间可能略晚于设置的超时时间。asyncio.timeout() 同样依赖协作式取消；被调用协程应该在 finally 中及时释放连接、文件和临时状态，不要吞掉 CancelledError。

从 Python 3.11 开始，这些 API 抛出内置的 TimeoutError。旧代码中常见的 asyncio.TimeoutError 已经不再是需要单独使用的异常类型。

7. 周期

asyncio 没有一个自动重复执行协程的 setInterval()。简单的周期任务通常使用循环和 asyncio.sleep() 实现：

fixed-delay.py
01
import asyncio

02


03
async def check_status():

04
    print("开始检查")

05
    await asyncio.sleep(0.3)

06
    print("检查完成")

07


08
async def poll():

09
    for _ in range(3):

10
        await check_status()

11
        await asyncio.sleep(1)

12


13
asyncio.run(poll())

这种方式是在一次检查完成后，再等待一秒。假设检查本身需要 0.3 秒，两次开始时间的间隔就接近 1.3 秒。这种调度方式叫固定延迟，适合不希望任务重叠的轮询。

如果希望任务尽量按照事件循环时钟上的固定时间点开始，就要根据目标时间计算每次等待长度：

fixed-rate.py
01
import asyncio

02


03
async def check_status():

04
    print("检查状态")

05
    await asyncio.sleep(0.3)

06


07
async def poll(interval, count):

08
    loop = asyncio.get_running_loop()

09
    next_run = loop.time()

10


11
    for index in range(count):

12
        await check_status()

13


14
        if index == count - 1:

15
            break

16


17
        next_run += interval

18
        delay = max(0, next_run - loop.time())

19
        await asyncio.sleep(delay)

20


21
asyncio.run(poll(interval=1, count=3))

这段代码会把任务耗时从下一次等待时间中扣除，因此开始时间更接近每秒一次。它仍然不是实时调度：如果某次检查超过一个周期，delay 会变成零，下一轮会尽快执行；它不会自动并行启动漏掉的任务。

两种方式没有绝对优劣：

方式开始间隔适合场景
固定延迟任务耗时加等待时间轮询、重试、不允许重叠的工作
固定频率尽量对齐预定时间点心跳、采样、界面刷新节奏

如果任务执行时间可能超过周期，需要明确选择跳过错过的周期、立即补跑，还是允许多个实例重叠。不要让这个行为由偶然的执行速度决定。

停止周期任务

无限循环应该放在 Task 中，并通过取消来停止。协程使用 finally 完成清理：

cancel-periodic.py
01
import asyncio

02


03
async def heartbeat():

04
    try:

05
        while True:

06
            print("发送心跳")

07
            await asyncio.sleep(1)

08
    finally:

09
        print("关闭心跳连接")

10


11
async def main():

12
    task = asyncio.create_task(heartbeat(), name="heartbeat")

13


14
    await asyncio.sleep(2.5)

15
    task.cancel()

16


17
    try:

18
        await task

19
    except asyncio.CancelledError:

20
        print("心跳任务已停止")

21


22
asyncio.run(main())

取消 Task 会中断它当前正在等待的 sleep()。调用方仍然要 await task，这样才能确认清理已经完成，并取走取消状态或其他异常。

周期任务中的异常也不能忽略。未处理异常会让这个 Task 直接结束，后续周期不再执行；如果业务需要重试，应只捕获明确可以恢复的异常，记录失败并应用退避策略，而不是用一个宽泛的 except 隐藏所有问题。

8. 边界

事件循环定时器适合进程内、允许随进程一起结束的短期调度，例如：

  • 等待一段时间后重试；
  • 限制网络请求或模型调用的最长耗时；
  • 每个服务进程各自维护心跳；
  • 定期刷新当前进程的内存缓存。

它不适合承担必须可靠执行的业务任务。进程重启、服务扩缩容或部署更新时，内存中的 TimerHandle 和 Task 都会丢失。

在 FastAPI 中还要考虑 worker 数量。如果启动四个 worker，每个进程都有自己的事件循环；在应用启动阶段创建一个周期 Task，通常会得到四份相同任务。对于「全系统只能执行一次」的账单生成、定时通知或数据归档，应使用具备持久化、分布式锁、失败重试和监控能力的任务队列或调度系统。

最后，定时器也不是解决阻塞问题的工具。回调到期后如果执行了耗时同步代码，仍然会阻塞事件循环。同步 IO 可以考虑 asyncio.to_thread()，CPU 密集型计算则更适合进程池或独立 worker。

正在验证登录状态
请稍候，验证完成后将继续显示文章内容
0
订阅