Python 与数据实践 · 第 19 章

事件循环

上一章的速查表里，"异步"一栏反复出现一个词：asyncio。而在 asyncio 的世界里，出场率最高的概念非事件循环（event loop）莫属——协程靠谁调度？await 为什么能"等而不卡"？答案全在它身上。本章…

约 11 分钟4,588 字9 个代码示例

上一章的速查表里，"异步"一栏反复出现一个词：asyncio。而在 asyncio 的世界里，出场率最高的概念非事件循环（event loop）莫属——协程靠谁调度？await 为什么能"等而不卡"？答案全在它身上。本章不看应用看原理，把这台"发动机"拆开来检查一遍。

1. 从一段"卡住"的代码说起

先回忆一下单线程世界的痛点（示意代码）：

def send_sms(phone: str, code: str):
    # 调用短信网关发送验证码
    # 请求发出去之后，线程就在这里干等，直到网关响应
    gateway.send(phone, code)  # 阻塞！


def write_log(msg: str):
    # 记一条日志，本来一毫秒就能搞定
    pass


send_sms("138****8000", "582014")  # 等网关的这几秒里……
write_log("验证码已发送")            # 这行只能干瞪眼

单线程里，代码从上到下排队执行：前面的任务被什么拖住了（网络慢、磁盘忙），后面的任务再轻也只能等。线程被死死按在原地，什么有用的事都做不了——这是对计算资源的纯浪费。

2. 异步到底是什么

异步是一种安排任务的方式：还是这一个线程，但不允许任何任务把线程占死。任务 A 发出请求、进入等待时，就把线程让出来，让任务 B、C、D 先跑；A 的结果回来了，再回来把 A 的后半段做完。

同步模式（一个任务从头占到尾）：
任务A │发请求│……干等……│处理结果│ 任务B │发请求│……干等……│处理结果│
      └──────────── 线程大把时间在发呆 ────────────┘

异步模式（等待时把线程让出去）：
任务A │发请求│     （等着，但线程去忙别的了）      │处理结果│
任务B        │发请求│      │处理结果│
任务C                │发请求│   │处理结果│
      └──────────── 线程几乎没闲着 ────────────┘

核心收益一句话：等待不再等于浪费。串行时被"干等"吞掉的时间，全部拿来跑别的任务了。

3. 异步和多线程的分界线

上一章的选型表已经给过结论，这里换个角度再推一遍：

计算密集型：CPU 从头忙到尾，根本没有"等待"可利用，单线程异步帮不上忙——多线程、多进程才是正解；
I/O 密集型：时间都耗在等外部响应上，异步用一个线程就能把等待时间全部填满。而且它比多线程省得多——线程要吃内存、切换要进内核，任务上万的规模线程扛不住，异步扛得住。

一句话：等待多、计算少，用异步；计算多、等待少，用线程或进程。

4. 事件循环：异步的发动机

在 Python 里，"事件循环"不是一个抽象概念，而是一个实实在在的对象——AbstractEventLoop 类的实例。它本质上是个不知疲倦的调度员：手里攥着任务队列，一圈一圈地转，每转一圈就检查一遍"谁到点了、谁的结果回来了、谁可以跑了"。

先认识它的几个基本操作（下面这段是"仪表盘讲解"，把 run_forever 那两行注释放开程序就会一直转下去，所以保持注释状态）：

import asyncio

# 造一个全新的事件循环对象
loop = asyncio.new_event_loop()

# 把它绑定到当前线程（一个线程同一时刻只挂一个循环）
asyncio.set_event_loop(loop)

# 取出当前线程绑定的循环
current = asyncio.get_event_loop()
print("当前线程的事件循环：", current)

# 解除绑定（对象还在，只是不再挂在这个线程名下）
asyncio.set_event_loop(None)

# 让循环转起来：从这一行开始，当前线程就交出去了
# 不撞南墙不回头——除非循环里有人调用 stop()，否则它永不返回
# current.run_forever()

# 喊停的方式
# current.stop()

两点提醒：

run_forever() 名副其实：没人调用 stop()，它就永远不返回，写在它后面的代码永远轮不到执行；
平时我们写 asyncio.run(...)，创建循环、运行、关闭这一整套动作都已被封装好。这里手动走一遍流程，是为了看清发动机本来的样子。

5. run_forever 的真面目

把 CPython 源码里 run_forever 的骨架捞出来，你会发现它短得惊人：

def run_forever(self):
    """Run until stop() is called."""
    while True:
        self._run_once()      # 转一圈：处理一轮任务
        if self._stopping:    # stop() 会把这个标记置为真
            break

一个 while True 加一个停止标记——所谓事件循环，就是一个"不停转圈"的死循环。真正的玄机，全在每一圈的 _run_once 里。

6. 每转一圈做什么：_run_once 的调度流程

_run_once 手里管着两条队列：

就绪队列（ready）：马上就能执行的回调；
延时队列（scheduled）：安排在将来某个时刻执行的回调，按到期时间排好序。

call_soon 直接放入 ──────►┌──────────────┐
                           │  ready 队列  │──本轮依次执行──► 跑回调
I/O 完成、延时到点 ───────►└──────────────┘
                                     ▲
call_later(delay, cb) 放入           │ 到点搬运
                           ┌──────────────┐
                           │  延时队列     │（按到期时间排序）
                           └──────────────┘

每一圈（即每一次 _run_once）的固定动作，按顺序是这五步：

搬运：翻一遍延时队列，把所有到期的回调搬进 ready；
算这一圈要等多久（timeout），分三种情况： - ready 里有活 → timeout = 0，一刻也不多等； - ready 空的、但延时队列有任务 → timeout = 队首到期时刻 - 当前时刻； - 两边都空 → timeout = None，无限期等下去；
等 I/O：带着 timeout 去监听 I/O。期间任何一个 I/O 完成，对应的回调立刻进 ready；若 timeout 耗尽还没有 I/O 动静，就回到第 1 步重新搬延时队列（此时大概率正好有任务到期了）；
换队列：把当前的 ready 交接出来，换上一条空队列——执行期间新塞进来的回调全部落到新队列里，留给下一圈；
执行：把交接出来的这批回调从头到尾跑一遍。跑完，这一圈结束，回到 run_forever 判断要不要继续转。

特别注意第 4 步这个细节：一圈只执行"开跑那一刻"已经就绪的回调。执行过程中新加入的（比如某个回调里又调了 call_soon）一律排到下一圈。把这个规则想透，后面预测代码输出时就不会掉坑。

7. 动手实验：call_soon 与 call_later

call_soon(cb) 把回调直接放进 ready；call_later(delay, cb) 把回调放进延时队列，delay 秒后进入 ready。亲手转一圈试试：

import asyncio

loop = asyncio.new_event_loop()


def immediate_task():
    print("我是 call_soon 放进 ready 的，很快轮到我")


def delayed_task():
    print("我是 call_later 安排的，1 秒后才进 ready")
    loop.stop()  # 干完这票就收工，让 run_forever 返回


loop.call_soon(immediate_task)
loop.call_later(1, delayed_task)

loop.run_forever()  # 转起来：先跑 immediate_task，约 1 秒后跑 delayed_task 并停下

print("循环已停止")

输出顺序：先是 immediate_task 的消息，隔约 1 秒是 delayed_task 的消息，最后是"循环已停止"。要是去掉 loop.stop()，程序会永远转下去——别忘了 run_forever 的脾气。

习题

一、预测下列代码的输出

import asyncio

loop = asyncio.new_event_loop()


def job1():
    print("任务一执行")


def job2():
    print("任务二执行")


def job3():
    print("任务三执行")


loop.call_soon(job1)
print("job1 已提交")
loop.call_soon(job2)
print("job2 已提交")
loop.call_soon(job3)
print("job3 已提交")

loop.run_forever()
print("全部结束")

提示：三句"已提交"发生在循环启动之前还是之后？run_forever 在什么条件下才会返回、让"全部结束"有机会打印？

二、预测下列代码的输出

import asyncio

loop = asyncio.new_event_loop()


def outer():
    print("A")
    loop.call_later(0, lambda: print("B"))
    loop.call_soon(lambda: print("C"))


def inner():
    print("D")
    loop.call_soon(lambda: print("E"))


loop.call_later(0, outer)
loop.call_soon(inner)

loop.run_forever()
print("done")

提示：延时 0 秒也是"延时"；回忆第 6 节的搬运顺序，以及"执行期间新加入的回调排到下一圈"的规则。

三、预测下列代码的输出

import asyncio

loop = asyncio.new_event_loop()


def step_one():
    print(1)


def step_two():
    print(2)
    loop.call_soon(lambda: print(3))
    loop.stop()


def step_three():
    print(4)


loop.call_soon(step_one)
loop.call_soon(step_two)
loop.call_soon(step_three)

loop.run_forever()
print("循环已停止")

提示：stop() 只是把停止标记置为真。当前这一圈已经交接出来的回调，还会不会执行完？print(3) 是排进这一圈还是下一圈？

可运行的版本见同目录的 demo.py——先预测，再运行对答案。

事件循环转起来了，但你也发现了：今天往里塞的都是普通函数，协程还没登场。协程是怎么被塞进循环、又是怎么被唤醒的？await 挂起后等到的那个"未来结果"（Future）到底是什么？下一章接着拆。

下一章：20 · Future 与协程