8.2 asyncio 异步编程

8.2 asyncio 异步编程

本节摘要：asyncio 用一个线程 + 一个事件循环 + 一群可暂停的任务，实现海量 IO 并发。本节讲事件循环的调度模型、async/await 的准确语义（await 是"让出控制权"的可暂停点）、gather 与超时取消，以及"async 传染性"与阻塞调用污染这两大陷阱的对策。

为什么还需要第三种并发

8.1 节的矩阵里，线程适合"IO 密集、几十个并发"。当并发到几百上千（万人同时连着的网关、抓万个页面的爬虫），线程就不划算了：每个线程默认栈 8MB 上限虽少、但调度与上下文切换的成本随数量线性上升，几千线程能明显压垮操作系统。

asyncio 的思路换了釜底抽薪的一招：既然 99% 的时间在等 IO，那就用一个线程轮询所有等待，谁就绪就让谁跑。等待不占线程，只占一个登记项——并发数从"线程数"变成"任务对象数"，万级起步。

事件循环：一个调度模型吃透

import asyncio

async def fetch(name, delay):
    print(f"{name} 发起请求")
    await asyncio.sleep(delay)      # 唯一的"暂停点"：让出控制权
    print(f"{name} 收到响应")
    return name

async def main():
    results = await asyncio.gather(
        fetch("A", 2), fetch("B", 1), fetch("C", 1.5)
    )
    print(results)

asyncio.run(main())
​

输出顺序值得细品：

A 发起请求
B 发起请求
C 发起请求      ← 三个任务都启动了
B 收到响应      ← 1 秒后，最早就绪的先恢复
C 收到响应
A 收到响应
['A', 'B', 'C']  ← gather 按提交顺序收集结果
​

单线程跑出了"同时等待、完成即续"的并发感。机制拆解：

1. asyncio.run 创建事件循环——一个反复执行"就绪队列取任务 → 跑到下一个暂停点"的循环；
2. async def 定义的是协程函数，调用它不执行、只造协程对象（与生成器 7.1 节"调用不执行"完全同构——await 的前身就是 yield from）；
3. await 声明"我要等一个可等待对象，期间让出线程"。事件循环趁机调度其他就绪任务；
4. gather 把多个协程包装成任务并发登记，全部完成后按序返回结果。

事件循环调度模型

await 的准确语义与"传染性"

三条必须校准的认知：

其一，协程不调度自己。 fetch("A", 2) 只是造了个对象，没有 await、没有 gather、没有 create_task 的话，它永远不会跑——经典新手 bug 是"调用了但日志毫无动静"。

其二，await 只能让出，不能创造并发。 顺序 await 三次 sleep(1) 总耗时 3 秒——这跟同步没区别。并发来自 gather/create_task：

async def slow_way():                       # 串行：约 3 秒
    await fetch("A", 1); await fetch("B", 1); await fetch("C", 1)

async def fast_way():                       # 并发：约 1 秒
    await asyncio.gather(fetch("A", 1), fetch("B", 1), fetch("C", 1))
​

其三，async 有传染性。 一个函数要 await，它自己就得是 async；调用它的也得 async……一路传到入口 asyncio.run。这是设计使然：暂停能力必须贯通调用链，断在中间就没法让出。工程对策是分层：IO 层 async 化，纯计算层保持普通函数，交界处用下一节的工具桥接。

超时、取消与并发原语

真实 IO 会挂。超时与取消是异步代码的必修课：

async def with_timeout():
    try:
        result = await asyncio.wait_for(fetch("slow", 10), timeout=2.0)
    except TimeoutError:                    # 3.11+ 就是内建 TimeoutError
        return "降级响应"

async def main():
    task = asyncio.create_task(fetch("B", 5))
    await asyncio.sleep(0.1)
    task.cancel()                           # 主动取消
    try:
        await task
    except asyncio.CancelledError:
        print("任务已被取消")               # 取消以异常形式送达，别裸吞
​

取消的机制值得敬畏：cancel 在任务下一个 await 点抛 CancelledError，任务应做好清理（6.2 节的 finally 语义在此完全适用）。同步原语 asyncio 也有一套镜像版：asyncio.Lock、asyncio.Semaphore、asyncio.Queue——虽然单线程免竞态，但"多任务交错访问共享缓存"这类逻辑竞态依然需要锁来保护临界区（await 点就是插队点）。

批量抓取的典型生产写法，用 Semaphore 限流防打爆对端：

async def bounded_fetch_all(urls, limit=10):
    sem = asyncio.Semaphore(limit)
    async def one(u):
        async with sem:
            return await fetch_page(u)
    return await asyncio.gather(*(one(u) for u in urls))
​

两大陷阱与桥接工具

陷阱一：阻塞调用污染。 在协程里调 time.sleep(3)、requests.get、重 CPU 循环——事件循环整个冻结，万级并发瞬间归零。自查口诀：异步代码里只允许出现 await 型等待。确实要用阻塞库时，把它扔进线程池代跑：

result = await asyncio.to_thread(blocking_io_fn, arg)   # 等待转交线程，循环照常转
​

陷阱二：连环 CPU。 即使全是 await，单线程的算力上限还是 100%。CPU 密集混合场景用 loop.run_in_executor 配进程池，回到 8.1 节的选型矩阵。

⚠️ 常见坑：忘记 await 协程（得到协程对象与一条 RuntimeWarning）；顺序 await 以为会并发；gather 中一个任务抛异常默认整体抛出（要收集全部成败用 return_exceptions=True）；Python 3.10 前后 asyncio.TimeoutError 与内建 TimeoutError 的名称统一问题（新版本已合并）。

💡 关键直觉：把事件循环当独木桥——桥上任何一步都得"要么飞快跑完、要么自觉让位（await）"。想在桥上打坐（阻塞）的，请去桥边的线程池。

本节要点回顾

• 模型：单线程事件循环 + 就绪队列 + 等待区；等待不占线程，万级并发靠登记项。
• await 三认知：协程不自动跑、await 不产生并发（gather 才产生）、传染性是设计而非缺陷。
• 超时与取消：wait_for / cancel 以异常形式送达，清理写 finally。
• 镜像原语：asyncio.Lock/Semaphore/Queue 防的是逻辑竞态（await 点插队）。
• 桥接：to_thread 接阻塞库，run_in_executor 接进程池。
• 选型闭环：IO 海量 asyncio、IO 少量线程、CPU 密集进程——8.1 节矩阵的最后一格填上了。

最后一章把视角拉到生态：数据分析、Web、爬虫、机器学习四大方向，以及它们如何复用全书这套机制底座。