asyncio 事件循环与协程：手写一个迷你 event loop

更新时间：2026-09-02。本文回答：async/await 到底是怎么"并发"的？一个线程怎么同时盯成千上万个连接？await 那一瞬间发生了什么，为什么有人说"别在协程里写阻塞调用"？事件循环（event loop）是不是什么魔法？ 00 · GIL、多进程与 asyncio 讲了 asyncio 的用法和 GIL 背景；这一篇往下挖一层，亲手用一百来行标准库写出一个能跑的事件循环（Future + Task + 定时器堆 + selectors/epoll），让你看完后再用 asyncio 时心里有一张完整的地图。本文 5 个 demo 全部在服务器（CentOS 7，Python 3.6.8）实测，代码见文末「代码位置」。

一、先校准一个认知：并发 ≠ 并行

asyncio 给你的是并发（concurrent），不是并行（parallel）。整个事件循环跑在一个线程上，任意时刻只有一段协程代码在执行。它快的秘诀不是"同时做多件事"，而是"绝不干等"——当一个任务要等 I/O（等网络数据、等超时）时，它主动让出 CPU，事件循环立刻切去跑别的任务；等 I/O 就绪了再回来。对 I/O 密集型服务（Web、爬虫、代理、网关），大部分时间本来就在等内核，协程把"等待"重叠起来，于是一个线程就能顶住海量连接。

对照 00 篇：GIL 让 Python 多线程在 CPU 密集场景无法真并行，要并行 CPU 就用多进程或下沉 C（见 06 · Python 与 C 互操作）；而 I/O 密集的高并发，协程 + 事件循环往往比多线程更省（没有线程栈和切换开销、没有锁）。协程靠的是协作式多任务（cooperative multitasking）——任务自己 await 让出，而不是被内核抢占。

那"协程能暂停、能恢复"这个能力从哪来？答案出人意料地朴素：生成器。

二、暂停与恢复的原语：生成器就是协程的底座

Python 的 async def 协程，底层的暂停/恢复机制和生成器（generator）是同一套。一个含 yield 的函数，调用时不执行函数体，而是返回一个生成器对象；每次 next() / send() 驱动它跑到下一个 yield 挂起，把值递出来；之后 send(x) 再让它从挂起处恢复，x 成为那个 yield 表达式的值。

demo1（d1_generator_sched/gen_sched.py）用这个原语手写一个轮转调度器：两个"任务"各跑 2 步，调度器交替 send 驱动它们：

after priming, yielded: ['A: step 0 done', 'B: step 0 done']
    [A] resumed with feedback='tick-1'
scheduler got: A: step 1 done
    [B] resumed with feedback='tick-2'
scheduler got: B: step 1 done
    [A] resumed with feedback='tick-3'
    [B] resumed with feedback='tick-4'
all coroutines finished

三个关键点：

1. 必须先"点火"（prime）：生成器创建后停在函数体第一行之前，第一次要用 send(None)（或 next()）把它推到第一个 yield。
2. yield 是让出点：任务跑到 yield 就把控制权还给调度器，调度器转而去 send 另一个任务——这就是协作式切换。
3. 任务结束的信号是 StopIteration：生成器函数 return 时抛 StopIteration，调度器据此把它移出队列。

这正是协程的全部魔法基础：能在指定点挂起、稍后从同一点带着数据恢复。async def/await 只是这套机制在语法和调度上的"官方封装"（yield from → await，普通生成器 → @coroutine/原生协程）。生成器基础见 intermediate 00 · 生成器与装饰器。

三、Future、Task 与最小事件循环

光有能挂起的协程还不够，还需要一个"管理者"决定什么时候唤醒谁。asyncio 的核心是三个概念，我们逐个手写：

• Future：一个"一次性结果占位符"。协程 await 一个还没完成的 Future 时挂起；Future 完成时（结果就绪），通知所有等它的回调。
• Task：把一个协程包起来，负责"驱动"它——协程每 yield 一个 Future，Task 就给这个 Future 注册一个回调：Future 一完成，就把协程再 send 一步。
• 事件循环（Loop）：维护两类等待——就绪队列（call_soon 排进来的、马上能跑的回调，一个 deque）和定时器堆（loop.sleep(sec) 排进来的、按到期时间排序的 Future，一个最小堆 heap）。循环反复：跑完所有就绪回调 → 把到期的定时器 Future 触发 → 没有就绪任务时，就阻塞等待下一个最早到期点。

demo2（d2_mini_loop/mini_loop.py）就是这个完整的迷你循环（约 60 行）。跑三个"睡 0.3s / 0.1s / 0.2s"的协程：

C: start, sleeping 0.30s
A: start, sleeping 0.10s
B: start, sleeping 0.20s
A: woke at t=0.10
B: woke at t=0.20
C: woke at t=0.30
total elapsed: 0.30s (concurrent max, not 0.6s sum)

三个 sleep 加起来本该 0.6s，实际只花 0.3s（取最大值）——因为它们是并发的：每个协程 yield loop.sleep(...) 后挂起，循环把三个到期点都放进定时器堆，到点依次唤醒。注意整个过程只有一个线程，没有任何"真等待"，循环靠堆顶的到期时间决定睡多久。loop.sleep 返回的就是一个 Future，它正是协程挂起和循环唤醒之间的"契约"。

四、接上 I/O 多路复用：selectors 就是 epoll

定时器只解决了"等时间"，真正的高并发要解决"等网络"。做法和定时器一模一样，只是 Future 的触发条件从"到点"换成"某个 socket 可读/可写"。Python 标准库 selectors 把操作系统的 I/O 多路复用封装成统一接口——Linux 上底层就是 epoll。

我们给迷你循环加一个 wait_io(sock, EVENT_READ/WRITE)：它注册一个 Future，把 socket 交给 selectors.DefaultSelector() 监听。循环在没有就绪回调时调用 sel.select(timeout)——这是整个循环唯一真正会阻塞的地方，但它是一次性盯着所有连接：任何一个 socket 就绪就返回。拿到就绪事件后触发对应 Future，那个协程就被唤醒去 accept/recv/send。Linux 上这个 selector 底层就是 epoll（见 epoll 原理 与 select/poll/epoll 对比）。

demo3（d3_epoll_echo/echo_server.py）就是这样一个单线程、epoll 驱动的 echo 服务器：主协程 yield 等待"监听 socket 可读"（有新连接），来了就 accept 并为每个连接建一个 Task；每个连接协程 yield 等"可读"（收到数据）再 yield 等"可写"然后回显。用 5 个并发客户端、每个发 3 条消息压它：

loop running for 3s, serving many clients on ONE thread...
echo server on 127.0.0.1:29567 (single-thread, epoll-driven)
echoed to ('127.0.0.1', 51032): b'c1-msg0'
echoed to ('127.0.0.1', 51030): b'c2-msg0'
echoed to ('127.0.0.1', 51034): b'c3-msg0'
...（5 客户端 × 3 消息全部正确回显，交错在一个线程上）
client ('127.0.0.1', 51032) disconnected
...
loop done
5 clients x 3 messages all echoed correctly by ONE event-loop thread

关键点：socket 全部设为非阻塞（setblocking(False)），所以 recv/send 没数据时不会卡住线程，而是抛 BlockingIOError 或我们先靠 wait_io 确保就绪。一个线程、一次 epoll_wait，同时服务所有连接——这就是 Node.js、Nginx、Redis 网络层、asyncio 的共同骨架。

一句话总结这个循环：协程负责"在该等的地方挂起"，事件循环负责"在能跑的时候唤醒"，而内核（epoll）负责"告诉你谁就绪了"。

五、真实 asyncio：换了个名字的同一套东西

我们手写的不是玩具类比，而是 asyncio 的真实架构。demo4（d4_asyncio_real/real_asyncio.py）直接把真实 asyncio 的内部对象打出来：

loop class: asyncio.unix_events._UnixSelectorEventLoop
selector   : EpollSelector
A: start, await asyncio.sleep(0.20)
B: start, await asyncio.sleep(0.10)
C: start, await asyncio.sleep(0.30)
B: done at t=0.10
A: done at t=0.20
C: done at t=0.30
elapsed 0.30s (max 0.3, not sum 0.6) -> concurrent, not parallel
coro type: coroutine

对照关系一目了然：

我们手写的	真实 asyncio
Future	asyncio.Future（loop.create_future()）
Task（包协程、逐步 send）	asyncio.Task（asyncio.ensure_future / create_task）
loop.sleep 往定时器堆排 Future	asyncio.sleep + 循环的 _scheduled 堆（call_later）
ready deque + call_soon	循环的 _ready 双端队列
wait_io + selectors	loop._selector（Linux 即 EpollSelector）+ sock_recv/add_reader
sel.select(timeout) 阻塞点	loop._run_once() 里的 selector.select
生成器协程 yield future	原生协程 await future（coro type: coroutine）

async def 定义出来的是一个 coroutine 对象（见最后一行），它和生成器一样有"挂起/恢复"状态机，只是用 await 而非 yield 挂起，并且只能被事件循环驱动。await future 的语义就是"如果 future 没完成就挂起，完成后拿它的结果恢复"。这就是为什么 05 · 字节码与求值循环 里说协程本质也是 code object + 一个可挂起的执行状态。

六、致命陷阱：协程里写阻塞调用，冻结整个循环

协作式多任务的代价是：只要你的协程不 await，就没人能抢过你。事件循环是单线程，一旦某个协程里调用了一个阻塞函数（time.sleep、同步的 requests.get、阻塞的文件读写、一个吃满 CPU 的长计算），在它返回之前，循环根本没机会去跑别的任务——整个进程"卡死"，所有其他协程的定时器、网络回调全部延迟。

demo5（d5_blocking_trap/blocking_trap.py）做对照：一个每 0.05s tick 的"心跳"协程，配一个睡 0.3s 的任务。坏版本用阻塞的 time.sleep(0.3)，好版本用协作的 await asyncio.sleep(0.3)：

=== BAD: blocking call inside coroutine ===
blocking_task: about to time.sleep(0.3) -- a BLOCKING call
blocking_task: done blocking
  fast[0] tick at 0.30
  fast[0] tick at 0.35
  fast[0] tick at 0.40

=== GOOD: await (cooperative yield) ===
async_sleep_task: awaiting asyncio.sleep(0.3) -- cooperative
  fast[1] tick at 0.00
  fast[1] tick at 0.05
  fast[1] tick at 0.10
async_sleep_task: done

bad  elapsed 0.45s (fast ticks stalled during the block)
good elapsed 0.30s (fast ticks interleaved with the wait)

证据很直接：坏版本里心跳第一次 tick 被拖到 0.30s（阻塞期间循环完全停转），好版本里心跳在 0.00/0.05/0.10 正常交错运行。这就是 asyncio 的头号戒律：

• 凡是等待，必须 await 一个异步操作（asyncio.sleep、aiohttp、asyncio.open_connection…），绝不在协程里直接调同步阻塞 API。
• 遇到不得不用的阻塞/重 CPU 代码，用 loop.run_in_executor 把它丢到线程池/进程池里跑，await 它的 future——这样循环线程不被占住。
• 这也解释了为什么异步生态要用 aiohttp 而不是 requests、要用 asyncpg 而不是 psycopg2：同步库内部的阻塞 socket 会让整个事件循环停摆。

七、交叉引用与延伸阅读

你想接着了解	去读
asyncio 的用法、何时选协程/多线程/多进程、GIL	00 · GIL、多进程与 asyncio
协程挂起/恢复的字节码与 code object 本质	05 · 字节码与求值循环
事件循环盯 socket 的内核机制：epoll 怎么工作	epoll 原理
select / poll / epoll 的演进与性能差异	I/O 多路复用对比
收发包时内核里发生了什么（socket 缓冲区）	socket 内核内幕
生成器 yield/yield from 基础	intermediate 00 · 生成器与装饰器
协程不是并行：CPU 密集仍需多进程或下沉 C	06 · Python 与 C 互操作
别的语言怎么实现协程/运行时	C++ 协程机制、Rust async 运行时

八、坑速查

• 协程里调阻塞函数（time.sleep、requests、阻塞文件 I/O、长 CPU 循环）：冻结整个事件循环，所有任务延迟。用异步库或 run_in_executor。
• 忘了 await：调用 async def 函数不 await 只会得到一个从不执行的 coroutine 对象（还会报 "coroutine was never awaited" 警告）。
• 自己创建的 Task 没保存引用：Task 可能被垃圾回收中途消失（和 ctypes 回调同类的生命周期坑）。把 Task 存进集合、done 后再丢弃。
• 阻塞库混用：在 asyncio 项目里用 requests/同步数据库驱动，等于白用 asyncio。网络层统一用异步驱动。
• 把协程当并行用在 CPU 密集场景：事件循环单线程，算得慢的部分不会被加速；CPU 密集用多进程（ProcessPoolExecutor）或下沉 C。
• 跨循环使用 future/task：Future 绑定在创建它的那个 loop 上，别把对象在不同事件循环间传来传去。
• run_until_complete vs run_forever：前者跑到给定 future 完成就停，后者跑到 stop()；服务器通常 run_forever。
• 异常被吞：Task 里的异常在你 await 它或取 task.exception() 之前不会抛出，未处理的任务异常容易"静默"，记得显式处理/回收。

九、代码位置

本文 5 个 demo 均在服务器（CentOS 7，Python 3.6.8）实测通过，纯标准库、无需编译，仓库路径 demos/python-expert/asyncio-loop/，直接 bash run_all.sh：

Demo	文件	演示内容
d1	d1_generator_sched/gen_sched.py	生成器 yield/send 的挂起-恢复原语，手写轮转调度器交替驱动两个协程
d2	d2_mini_loop/mini_loop.py	手写 Future/Task/Loop（就绪 deque + 定时器 heap），三个 sleep 并发取 max 0.3s
d3	d3_epoll_echo/echo_server.py + echo_client.py	迷你循环接 selectors（epoll），单线程非阻塞 echo 服务器，5 客户端 × 3 消息并发回显
d4	d4_asyncio_real/real_asyncio.py	真实 asyncio 内部：_UnixSelectorEventLoop + EpollSelector，与手写版逐项对照
d5	d5_blocking_trap/blocking_trap.py	阻塞 time.sleep 冻结循环（心跳拖到 0.30s）vs await asyncio.sleep 协作交错

服务器上：cd /tmp/asyncio-loop && bash run_all.sh（d3 会后台起服务器、跑客户端、3 秒后自动停）。

十、动手练习

1. 在 d2 的迷你循环上加一个 call_later(delay, fn) 方法（提示：复用定时器堆，到点把 fn 塞进 ready 队列），验证它能和 sleep 协程共存。
2. 给 d3 的 echo 服务器加一个"每个连接最多回显 N 条就主动关闭"的逻辑，观察连接协程 return 后 Task 自然结束。
3. 在 d5 的坏版本里把 time.sleep(0.3) 换成 await loop.run_in_executor(None, time.sleep, 0.3)，验证心跳是否恢复交错——理解 executor 如何"绕开"阻塞。
4. 用 asyncio.gather 并发抓 3 个 URL（用 aiohttp 或 asyncio.open_connection 手写 HTTP 请求），对比串行 requests 的总耗时，解释为什么总耗时≈最慢的一个。
5. 阅读标准库 asyncio.base_events.BaseEventLoop._run_once 的源码，对照本文 PlantUML 流程图，指出"就绪回调"和"selector.select"在真实代码里分别对应哪几行。