1. 职责

协程能够暂停和恢复，但协程本身不知道应该在什么时候恢复，也不知道暂停期间应该运行谁。事件循环（event loop）就是负责这项工作的调度器。

假设程序中有三个任务：

• 任务 A 正在等待 HTTP 响应；
• 任务 B 正在等待数据库结果；
• 任务 C 已经准备好继续运行。

事件循环不会一直守着任务 A。它会先运行已经就绪的任务 C，同时监听任务 A 和任务 B 等待的 IO。某个 IO 完成后，对应任务重新获得运行资格，事件循环再选择合适的时机恢复它。

事件循环主要负责：

事件循环不是线程池，也不是让代码自动并发的语法。通常一个事件循环运行在一个操作系统线程中，这个线程在同一时刻只执行一个 Task 或回调。任务必须运行到 await、结束或抛出异常，事件循环才有机会运行其他任务。

因此，asyncio 采用的是协作式调度：任务主动在可等待的位置让出执行权，而不是由事件循环在任意一行代码上强制打断任务。

2. 结构

理解事件循环之前，需要先看清它周围的几个对象：

它们之间的关系可以简化为：

01
协程对象
02
  ↓ 包装
03
Task
04
  ↓ 进入就绪状态
05
事件循环运行 Task
06
  ↓ 遇到尚未完成的 await
07
Task 暂停，等待 Future 或 IO
08
  ↓ 等待结果完成
09
Task 重新进入就绪状态
10
  ↓
11
事件循环恢复 Task

Task 是事件循环真正调度的协程执行单元。Future 更像结果占位符：结果没有准备好时，等待它的 Task 暂停；Future 完成后，会安排等待它的 Task 重新进入就绪状态。

事件循环不会不停遍历所有暂停的协程，询问它们是否完成。IO 监听器、定时器和 Future 会在条件满足时通知事件循环，把相关工作重新放回可运行的集合中。

「就绪队列」「定时器集合」和「IO 监听器」适合用来建立概念模型。不同操作系统和事件循环实现的内部数据结构可能不同，业务代码不应该依赖它们的具体实现细节。

3. 调度

一次事件循环迭代可以大致理解为以下步骤：

1. 检查哪些定时任务已经到期；
2. 计算最多可以等待 IO 多久；
3. 向操作系统查询哪些 IO 已经就绪；
4. 把到期定时器、IO 回调和其他已就绪工作加入待执行集合；
5. 依次运行这些 Task 或回调；
6. Task 遇到尚未完成的 await 后暂停，事件循环继续处理其他工作；
7. 开始下一次迭代。

这是便于学习的简化过程，不是要求应用开发者手写的循环。平时使用 asyncio.run()，Python 会负责驱动事件循环。

下面可以观察两个 Task 如何交替执行：

01
import asyncio
02

03
async def worker(name):
04
    print(f"{name}：开始")
05
    await asyncio.sleep(0)
06
    print(f"{name}：恢复")
07

08
async def main():
09
    print("main：创建任务")
10
    first = asyncio.create_task(worker("任务一"))
11
    second = asyncio.create_task(worker("任务二"))
12

13
    print("main：等待任务")
14
    await asyncio.gather(first, second)
15
    print("main：结束")
16

17
asyncio.run(main())

通常可以观察到下面的顺序：

1
main：创建任务
2
main：等待任务
3
任务一：开始
4
任务二：开始
5
任务一：恢复
6
任务二：恢复
7
main：结束

create_task() 把两个协程包装成 Task，并安排它们运行。main() 执行到 await gather(...) 后暂停，事件循环开始推进两个子任务。

示例中的 await asyncio.sleep(0) 不是真的等待时间，而是主动暂停当前 Task，让事件循环有机会运行其他就绪任务。它适合演示调度，不应该被当成修复阻塞代码的通用办法。

协作式调度意味着公平性不是自动保证的。如果一个 Task 长时间计算且没有遇到 await，它会一直占用事件循环线程：

1
async def calculate():
2
    total = 0
3
    for number in range(50_000_000):
4
        total += number
5

6
    return total

虽然 calculate() 使用了 async def，但函数内部没有真正的等待点。它开始运行后，其他 Task 只能等它计算结束。

4. IO

IO 是 Input/Output（输入/输出）的缩写。在异步服务中，它通常表示等待网络、数据库、文件或其他外部资源。socket 可以先理解为程序与网络连接交互的对象。

异步 IO 的关键不是反复询问网络请求有没有完成，而是把等待交给操作系统。

以异步 HTTP 请求为例，可以把过程理解为：

1. HTTP 客户端创建非阻塞 socket；
2. socket 暂时还不能读取完整响应；
3. 客户端创建一个等待结果，并把 socket 注册给事件循环；
4. 当前 Task 在 await 位置暂停；
5. 事件循环继续运行其他 Task；
6. 操作系统发现 socket 已经可以读取；
7. 事件循环收到 IO 通知，执行对应处理逻辑；
8. 等待结果完成，原 Task 重新进入就绪状态；
9. 事件循环恢复原 Task。

底层通常依赖操作系统提供的高效 IO 机制，例如 Linux 的 epoll、macOS 的 kqueue 和 Windows 的 IOCP。具体实现会因平台与 Python 版本而不同，但核心目标都是让一个线程同时管理大量处于等待状态的连接。

所以，异步 HTTP 通常不是「一个 HTTP 请求对应一个线程」，也不需要事件循环额外创建一个「HTTP 线程」。事件循环所在的线程负责运行 Python 回调和 Task，操作系统负责监控大量 IO 状态。

定时等待也是类似的思路：

1
import asyncio
2

3
async def remind():
4
    print("开始等待")
5
    await asyncio.sleep(2)
6
    print("恢复执行")
7

8
asyncio.run(remind())

asyncio.sleep(2) 会注册定时等待并暂停当前 Task，通常不会创建一个专门的定时器线程。两秒后只表示这个 Task 具备恢复资格；如果事件循环此时正在执行其他代码，实际恢复时间可能稍晚。

5. 回调

大多数业务代码只需要协程和高层 API，但事件循环也提供了调度普通回调的低层接口：

01
import asyncio
02

03
def ready_callback():
04
    print("就绪回调")
05

06
def timer_callback():
07
    print("定时回调")
08

09
async def main():
10
    loop = asyncio.get_running_loop()
11

12
    ready_handle = loop.call_soon(ready_callback)
13
    timer_handle = loop.call_later(1, timer_callback)
14

15
    print(type(ready_handle).__name__)
16
    print(type(timer_handle).__name__)
17
    await asyncio.sleep(1.2)
18

19
asyncio.run(main())

call_soon() 返回 Handle，call_later() 返回 TimerHandle。只要回调尚未执行，就可以通过它们的 cancel() 方法取消调度。

call_later() 使用相对延迟，call_at() 使用与 loop.time() 相同的单调时钟。单调时钟只适合计算时间间隔，不应该把 loop.time() 当成日期时间戳。

这些回调是普通同步函数。回调一旦开始运行，在返回之前会占用事件循环线程，因此回调内容也应该保持短小，不能直接执行阻塞 IO 或长时间计算。

大部分 asyncio 对象都不是线程安全的。在事件循环线程之外调用 call_soon() 并不安全，必须使用 call_soon_threadsafe()；从其他线程提交协程时，则使用 run_coroutine_threadsafe()。

Python asyncio 没有 JavaScript 那种公开且严格对应的「宏任务队列」和「微任务队列」。call_soon()、Task 恢复、Future 完成和定时回调都遵循 asyncio 自己的调度规则，不应该机械套用 Promise.then() 与 setTimeout() 的优先级结论。

6. 运行

普通 Python 程序应该优先使用 asyncio.run() 作为异步入口：

01
import asyncio
02

03
async def main():
04
    loop = asyncio.get_running_loop()
05
    print(loop.is_running())  # True
06
    await asyncio.sleep(0.1)
07
    return "完成"
08

09
result = asyncio.run(main())
10
print(result)

asyncio.run() 会完成一整套生命周期管理：

1. 创建事件循环；
2. 把入口协程作为主任务运行；
3. 驱动事件循环直到入口任务完成；
4. 完成异步生成器等清理工作；
5. 关闭默认执行器；
6. 关闭事件循环；
7. 返回入口协程的结果，或者向外抛出异常。

asyncio.run() 不能在同一线程已经运行事件循环时再次调用。FastAPI、异步测试框架和部分交互式环境已经管理事件循环，此时应该直接 await：

1
async def load_data():
2
    return "数据"
3

4
async def endpoint():
5
    return await load_data()
6

7
# 不要在 endpoint() 中调用 asyncio.run(load_data())

在协程和回调内部需要取得当前事件循环时，优先使用 asyncio.get_running_loop()。旧教程经常展示 get_event_loop()、new_event_loop()、run_forever() 和手动 close()，这些是低层控制接口，不是普通应用的首选写法。事件循环 policy 体系也已经弃用，因此不建议新项目把 policy 配置作为主线方案。

事件循环不能解决同步阻塞代码。下面的接口会让同一事件循环上的其他 Task 一起等待：

1
import time
2

3
async def bad_task():
4
    time.sleep(3)
5
    return "完成"

优先把同步库替换为异步库。确实无法替换的短时间阻塞 IO，可以使用 asyncio.to_thread() 隔离：

01
import asyncio
02
import time
03

04
def blocking_work():
05
    time.sleep(1)
06
    return "完成"
07

08
async def main():
09
    result = await asyncio.to_thread(blocking_work)
10
    print(result)
11

12
asyncio.run(main())

开发阶段还可以启用调试模式，帮助发现忘记等待的协程、错误的跨线程调用和执行时间过长的回调：

1
import asyncio
2

3
async def main():
4
    await asyncio.sleep(0.1)
5

6
asyncio.run(main(), debug=True)

调试模式会增加额外开销，适合开发和排查问题，不应在不了解成本的情况下直接作为高负载生产环境默认配置。