Python 与数据实践 · 第 20 章

Future 与协程

第 19 章把事件循环拆开看了个明白：一个线程、两个队列、一串回调，循环每一圈都在做"执行 ready 队列里排队的函数"这件事。但有两个问题当时刻意绕开了：

约 23 分钟9,136 字17 个代码示例

那些"正在等待中"的任务，在代码里用什么表示？
回调写多了，代码会不会散成一地碎片？

这两个问题的答案分别是 Future 和协程。搞懂它们，async / await 这套语法就不再是一团迷雾，而是一个你能亲手拆装的机器。

1. Future：一张"尚未兑付的凭据"

先想一个日常场景：你在奶茶店下单，店员递给你一张取餐凭据。此刻奶茶还没做出来，但凭据保证了两件事——将来一定会兑付，兑付时一定有个说法。Future 就是代码世界里的这张凭据：它代表一个已经开始、但要等到将来才能有结果的任务。

每一张凭据只有两种状态：

未完成：还在等待，结果尚不存在；
已完成：有了说法。说法又分三种——
正常完成：拿到了结果；
出错完成：拿到了一个异常；
被取消：任务中途撤销，等不到结果了。

1.1 先造一张凭据出来

Future 对象一般不直接实例化，而是通过事件循环的工厂方法创建：

import asyncio

loop = asyncio.new_event_loop()
fut = loop.create_future()

print(fut.done())  # False —— 还没完成
print(fut.result())  # 抛出 InvalidStateError：结果根本还不存在

注意 result() 的行为：凭据未兑付时去取结果，不是返回 None，而是直接抛 InvalidStateError——"没有结果"和"结果是空"是两码事。

1.2 三种"兑付"方式挨个试

正常完成——用 set_result 兑付：

fut.set_result("报表已生成")

print(fut.done())    # True
print(fut.result())  # 报表已生成

出错完成——用 set_exception 兑付一个异常：

fut2 = loop.create_future()
fut2.set_exception(ValueError("数值不合法"))

print(fut2.done())       # True
print(fut2.exception())  # ValueError: 数值不合法
print(fut2.result())     # 抛出 ValueError：取结果时异常被重新抛出

取消——用 cancel 直接作废：

fut3 = loop.create_future()
fut3.cancel()

print(fut3.done())        # True —— 取消也算一种"完成"
print(fut3.cancelled())   # True
print(fut3.result())      # 抛出 CancelledError
print(fut3.exception())   # 同样抛出 CancelledError

三种结局殊途同归：done() 都变成 True，此后凭据定格，状态不再改变。

1.3 兑付了再叫我：add_done_callback

更多时候我们不想守着凭据反复问"好了没"，而是留个话："好了叫我"。这就是 add_done_callback：

fut4 = loop.create_future()


def on_ready(f: asyncio.Future) -> None:
    try:
        print(f"凭据兑付，结果：{f.result()}")
    except Exception as exc:
        print(f"凭据兑付，但出错了：{exc}")


fut4.add_done_callback(on_ready)

有个细节值得停一下：注册回调这一步什么都不会发生。回调不会立刻执行，而是等 fut4 完成后，被投递到事件循环的 ready 队列里排队，由循环统一调度执行——这正是第 19 章讲过的调度机制。Future 负责的是"完成任务"的表达，通知和执行永远由事件循环经手，两者职责分得清清楚楚。

2. 用 Future 改造两段经典等待

空讲 API 有点飘，来看 demo 里两个真实封装。本章 demo 目录下有三个文件：alarm.py（延时封装）、probe.py（网络请求封装）、runner.py（启动器）。

2.1 闹钟：把"等 N 秒"封装成 Future

demo/alarm.py 的核心只有四行：

def ring_after(seconds: float) -> asyncio.Future:
    loop = asyncio.get_event_loop()
    fut = loop.create_future()
    loop.call_later(seconds, fut.set_result, "叮——")
    return fut

读懂它：call_later 是第 19 章的老朋友——seconds 秒后，它会带着参数执行 fut.set_result("叮——")，凭据随之兑付。调用方的体验是：拿到凭据、注册回调、转身走人，等待期间线程完全空闲。

配套的 main 部分注册了一个回调，两秒后闹钟响起时触发——而"闹钟已设定"这行代码是立刻打印的，等待没有卡住任何人。

2.2 拨测：把一次 HTTP 请求封装成 Future 链

demo/probe.py 是本章最有嚼头的代码，它用非阻塞 socket 完整实现了一次 HTTP 拨测，分三层：

open_link(host, port)：发起非阻塞连接。非阻塞的 connect 几乎必然先抛 BlockingIOError（请求已发出、结果未到），真正的"连上了没有"要靠 loop.add_writer 注册的回调来回答——描述符变得可写，就说明连接有了最终结果，此时再检查 SO_ERROR 区分成功与失败；
read_reply(sock, request)：把请求发出去，用 loop.add_reader 挂上读取回调，循环接收响应，直到收到空字节 b""——那是对端关闭连接的信号，读取到此为止；
probe_http(host, port, path)：把前两层串起来，对外只暴露一个总 Future——连接 Future 完成后触发读取，读取 Future 完成后兑付总结果，错误也顺着链条向总 Future 传递。

一个 Future 兑付触发下一段流程的开启，这就是纯 Future 风格的异步组织方式。建议对照文件里的注释逐行读通。

2.3 启动器：runner.py

前两个 demo 都需要一个"把循环转起来"的入口，demo/runner.py 就是干这个的：

def boot(entry) -> None:
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.call_soon(entry)
    loop.run_forever()

新建循环、绑定线程、把入口函数塞进 ready 队列、转起来——四步正是第 19 章手动操作事件循环的标准姿势。

3. 回调的裂缝：代码开始向右生长

Future 解决了"结果何时到位"的表达问题，但看看拿到结果之后的代码长什么样：

ring_after(2).add_done_callback(lambda _: print("两秒到了"))

单看还行。可异步任务的典型形态是"完成后紧接着下一件事"，比如拨测完一个站点再拨测下一个、三个站点全部测完再汇总——用回调写出来就是这样：

probe_http("example.com").add_done_callback(lambda r1:
    probe_http("example.org").add_done_callback(lambda r2:
        probe_http("example.net").add_done_callback(lambda r3:
            print("三个站点都拨测完了")
        )
    )
)

这就是名声在外的回调地狱：代码每多做一步就向右缩进一层，业务只有三步，缩进已经三层。更糟的还在后面——

想用 try 统一处理三步里任何一步的异常？对不起，异常发生在回调里，try 根本罩不住；
想在第三步用到第一步的中间变量？闭包套闭包，越套越深；
整段代码的执行顺序被切碎埋进了一个个匿名函数，从上往下读代码已经读不出业务流程。

我们需要一种魔法：写的时候是从上往下的顺序，执行的时候还是异步不阻塞。协程就是这门魔法。

4. 协程：把异步代码摆回"从上往下"

4.1 async def 与协程对象

在函数定义前加一个 async，它就成了协程函数：

async def double_check() -> str:
    ...

调用协程函数，得到的不是执行结果，而是一个协程对象——函数体一行都没跑，只是把"这段将要执行的流程"打包了起来。这和第 11 章生成器函数调用后得到生成器对象、函数体同样不执行，是一模一样的脾气。

4.2 await：在这里停一下，等结果回来

协程对象内部，用 await 等一个结果。把前面那个三层回调地狱改写成协程：

import asyncio

from alarm import ring_after
from probe import probe_http


async def patrol() -> str:
    first = await probe_http("example.com")
    print("第一站完成，响应头", len(first.split("\r\n\r\n")[0]), "字符")
    second = await probe_http("example.org")
    print("第二站完成，响应头", len(second.split("\r\n\r\n")[0]), "字符")
    await ring_after(1)
    print("歇了一秒")
    third = await probe_http("example.net")
    print("第三站完成，响应头", len(third.split("\r\n\r\n")[0]), "字符")
    return "巡检结束"

（示例在 demo 目录下运行，alarm 和 probe 都是同目录模块。）

读起来完全是同步代码的样子：第一站、第二站、歇一秒、第三站，从上往下。但执行起来，await 所在之处，协程会挂起——把线程让出来还给事件循环，事件循环趁这段时间去推进别的协程；被等待的对象一旦完成，协程就从断点处恢复，拿到结果继续往下跑。

几条规则记牢：

await 只能出现在协程函数（async def）里；
await 后面必须跟一个 awaitable 对象，最常见的三种：协程对象、Task、Future——上面代码里 probe_http 和 ring_after 返回的都是 Future；
协程对象如果不被 await、也没人驱动它，就只是一段打包好却从未执行的代码。

5. 协程和线程，到底差在哪

"一个线程里跑很多任务"这件事，第 18 章用线程也能做（每个任务开一个线程等着）。那协程存在的意义是什么？逐项对比：

对比维度	🧵 线程	🍃 协程
谁来调度	操作系统内核，什么时候切走你说了不算	事件循环（如 asyncio），调度权在应用层手里
资源开销	高：每个线程独享一块 MB 级的栈，外加内核对象，创建、销毁都不便宜	低：一个协程只要 KB 级的小栈，说建就建
切换成本	贵：一次切换要在用户态和内核态之间往返，烧掉数百个时钟周期	便宜：纯用户态换一下寄存器上下文，几十个周期搞定
数据共享	共享内存意味着随时可能互踩，锁、信号量一个不能少，死锁和竞态条件防不胜防	默认单线程内串行执行，同一时刻只有一个协程在跑，天然没有"抢跑道"（多线程事件循环另说）
利用多核	✅ 天生支持，线程可以被摊到多个核心并行计算	❌ 单个事件循环只占一个核，想吃满多核要靠 `run_in_executor` 或多进程
拿手的活	计算密集型任务；对响应速度要求苛刻的 I/O	海量 I/O 等待型任务：数据采集、高并发接口、聊天长连接，开几万个协程面不改色
代表技术	`threading`、`ThreadPoolExecutor`	`asyncio`（`async`/`await` 语法）、`trio`、`gevent`
数量级	上百个线程，开销就已经很可观	几万、几十万个协程都算日常

一句话收束：线程靠人海（多核并行），协程靠周转（单核高频调度）。等待满天飞的场景，协程用极低的成本把等待时间全部填满，这是线程给不了的性价比。

6. 亲手推一次协程

现在回答一个关键问题：协程对象自己不会动，那"驱动"到底发生了什么？我们手动推一次，看清里面的机关。

协程对象和生成器一样支持 send：coro.send(None) 把协程发动起来，它会一路跑到第一个 await 处停下，并把正在等待的那个对象（通常是个 Future）作为 send 的返回值交出来。之后每当这个 Future 完成，就把结果 send 回去，协程便从断点恢复、继续跑向下一段；协程彻底跑完时，send 会抛出 StopIteration，return 的值就挂在异常的 value 上——又是第 11 章生成器的老规矩。

据此可以写一个通用的手动驱动器：

def drive(coro):
    done = loop.create_future()          # 用来装协程的最终返回值
    current = coro.send(None)            # 发动协程，拿到第一个 await 的 Future

    def resume(fut: asyncio.Future) -> None:
        nonlocal current
        try:
            current = coro.send(fut.result())   # 结果送回去，协程恢复执行
            current.add_done_callback(resume)  # 新的等待点，继续挂回调
        except StopIteration as e:
            done.set_result(e.value)           # 协程跑完了，return 值在 e.value

    current.add_done_callback(resume)
    return done

配合事件循环跑起来（probe 里用了 get_running_loop，所以驱动这活儿要安排在循环运行期间进行）：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)


def launch() -> None:
    fut = drive(patrol())

    def on_over(f: asyncio.Future) -> None:
        loop.stop()
        print("协程的返回值：", f.result())

    fut.add_done_callback(on_over)


loop.call_soon(launch)
loop.run_forever()
print("结束")

看清楚这套机关的本质：协程加 Future，正好拼成一个自我延续的回调链。协程每 await 一次，就产出一张新凭据，凭据兑付的回调负责唤醒协程——上一节"向右生长的回调地狱"，被协程在垂直方向上折叠了起来。章节根目录的 demo.py 把这个手动驱动做成了可运行版本，跑一遍、对照输出读一遍，胜过看十遍文字。

7. Task：官方出品的协程驱动器

上面 drive 函数干的事，官方早就标准化了——Task。Task 是 Future 的子类，专职驱动协程对象：

def launch() -> None:
    task = asyncio.create_task(patrol())  # 协程对象交给官方驱动器

    def on_over(t: asyncio.Task) -> None:
        loop.stop()
        print("协程的返回值：", t.result())

    task.add_done_callback(on_over)


loop.call_soon(launch)
loop.run_forever()
print("结束")

和手动版输出完全一致，但 drive 那十几行全没了。asyncio.create_task 做了三件事：

把协程对象包成一个 Task；
立即把这个 Task 排进事件循环的 ready 队列——协程马上就会开始执行；
协程 return 时 Task 完成，返回值就是 Task 的结果；协程抛异常，Task 也随之以异常收场。

因为 Task 本身是 Future，所以它天然可以 await，也可以 add_done_callback——底层协议一以贯之。

8. asyncio.run：一步到位

日常写代码，连"建循环、绑线程、call_soon"这套仪式都可以省掉。asyncio.run 接收一个协程对象，内部自动完成"创建事件循环 → 包装成 Task → 跑到完成 → 拿到返回值 → 关闭循环"的全流程：

result = asyncio.run(patrol())
print(result)
print("结束")

一行顶前面一屏。前面手动操作循环的代码不是白学——它们是 asyncio.run 的内部实现缩影，理解了缩写，才明白这一行在替你做什么。

9. 把这一章缝起来

async def 定义的是协程函数，调用它得到协程对象，函数体此时并未执行；
awaitable 有三个常客：协程对象、Task、Future；await 只能写在协程函数里，它让当前协程挂起、让出线程，等结果到位再从断点恢复；
Future 是底层的"结果凭据"，两种状态、三种完成结局（正常、异常、取消），完成通知靠 add_done_callback 投递进 ready 队列；
Task 是 Future 的子类，是官方的协程驱动器：create_task 之后协程立即进 ready 队列开跑，协程的 return 值就是 Task 的结果；
asyncio.run(coro) 是日常入口，内部把协程转成 Task 驱动到完成。

下一章我们放下"造轮子"，全面启用官方和社区 API 做真正的实战。

作业

一、读懂两段封装

通读 demo 目录下的 alarm.py 和 probe.py（配套入口是 runner.py），读完后应该能脱稿回答：

非阻塞 socket 的 connect 为什么几乎必然先抛 BlockingIOError？靠什么机制得知连接的最终结果？
读取响应时，收到 b"" 意味着什么？为什么不在这时再 recv 了？
probe_http 内部三层 Future 是怎么环环相扣的？某一层出错时，错误如何传到最外层？

二、实现 `collect_all` 函数

并发驱动多个协程，等它们全部完成后，把各自的结果按传入顺序打包进一个 Future 返回——功能对标 asyncio.gather：

import asyncio
from typing import Coroutine

from alarm import ring_after


def collect_all(*jobs: Coroutine) -> asyncio.Future:
    # 你的代码
    pass


async def pull(sensor: str, seconds: float) -> str:
    await ring_after(seconds)
    return f"{sensor} 读数就绪"


async def main() -> None:
    outcomes = await collect_all(
        pull("温度计", 2),
        pull("湿度计", 1),
        pull("气压计", 3),
    )
    print(outcomes)  # 预期: ['温度计 读数就绪', '湿度计 读数就绪', '气压计 读数就绪']


asyncio.run(main())

提示： 每个协程都用 create_task 推上跑道；结果按各自下标放回列表（完成顺序和返回顺序不是一回事）；用计数器数齐了再兑付总 Future。

三、实现 `Gate` 类

一道只能开合一次的闸门——功能对标 asyncio.Event：

import asyncio

from alarm import ring_after
from collect import collect_all


class Gate:
    # 你的代码
    pass


async def demo() -> None:
    gate = Gate()

    async def audience() -> None:
        print("观众: 进入直播间，坐等开播")
        await gate.wait()
        print("观众: 开播了！")

    async def host() -> None:
        print("主播: 正在调试设备……")
        await ring_after(1)
        gate.lift()
        print("主播: 各位晚上好，正式开播")

    await collect_all(audience(), host())


asyncio.run(demo())

# 预期输出：
"""
观众: 进入直播间，坐等开播
主播: 正在调试设备……
主播: 各位晚上好，正式开播
观众: 开播了！
"""

提示： wait 里现场创建一个 Future 并 await 它；lift 做的事就是兑付这个 Future。想知道自己的实现和官方差在哪，可以去读 asyncio.Event 的源码——它用等待者列表支持多个等待方，参考实现 demo/gate.py 则是只保证单个等待方的简化版。

下一章：../21-asyncio实战/课件.md