Python 异步编程

asyncio 用事件循环调度协程，在 await 点协作切换以重叠 I/O；分清并发与并行，并处理阻塞、超时与取消边界。

#type / concept #status / growing #tech / dev #resource / python #tech / lang / python

[!info] 关联笔记

所属 MOC：并发与异步 · 学习路线 · Python MOC

前置概念：函数与装饰器 · 错误处理 · 并发模型总览 · GIL 与线程

后续概念：任务、超时与取消 · 同步原语与队列

应用：FastAPI · 大模型应用开发 · HTTP 客户端

容易混淆：多线程并行、多进程 CPU 加速、用 sleep 假装同步

Python 异步编程

这个概念为什么出现

现代后端与 AI 编排大量时间花在等网络/磁盘：拉模型 API、查向量库、扇出多个上游。若串行等待，尾延迟是各段之和；若为每个连接开系统线程，高并发下内存与调度成本又很高。

asyncio 给出第三种主路径：

用协程表达“算一会 → 等待 → 再算”
用 事件循环 在等待点切换到其他就绪任务
在单线程内重叠大量 I/O

它不自动提供多核 CPU 并行。若协程里调用阻塞 SDK，或把重计算塞进循环线程，所有任务都会一起卡住。

[!abstract] 一句话理解 async def 定义协程函数；await 在可等待对象上让出执行权；事件循环调度多个任务重叠 I/O。asyncio 解决并发等待结构，不把 CPU 密集自动变成多核并行。

最小可运行示例

先把示例放进业务场景，再看代码：

场景：价格聚合服务同时查询三个上游报价

比价中台要对 A/B/C 三个上游拉报价，取最低价返回。
串行 await A; await B; await C 会把 RTT 相加。
正确做法：三个请求并发发出，等最慢的那个结束（此处用 asyncio.sleep 模拟网络 RTT，不是用来当锁）。

# price_fanout_async.py
# 业务意图：并发拉取多上游报价并汇总最低价。
# 教学点：
# - async def / await；
# - asyncio.gather 并发；
# - asyncio.run 应用入口；
# - 模拟 I/O 必须 await asyncio.sleep，禁止 time.sleep 堵循环。

from __future__ import annotations

import asyncio
from time import perf_counter


async def fetch_quote(source: str, delay_s: float, price: int) -> tuple[str, int]:
    # 业务：请求某个上游报价；delay 表示网络往返。
    print(f"start {source}")
    await asyncio.sleep(delay_s)  # 让出事件循环，其他 fetch 可前进
    print(f"done {source}")
    return source, price


async def best_quote() -> tuple[str, int]:
    # gather：并发调度三个协程，全部完成后返回结果列表。
    results = await asyncio.gather(
        fetch_quote("A", 0.30, 120),
        fetch_quote("B", 0.20, 110),
        fetch_quote("C", 0.25, 115),
    )
    # 业务规则：选最低价
    return min(results, key=lambda item: item[1])


def main() -> None:
    t0 = perf_counter()
    source, price = asyncio.run(best_quote())
    elapsed = perf_counter() - t0
    print(f"best={source}:{price} elapsed≈{elapsed:.2f}s")


if __name__ == "__main__":
    main()

建议运行：

python price_fanout_async.py

期望输出（完成顺序可能交错，总耗时约 0.30s 量级而非 0.75s）：

start A
start B
start C
done B
done C
done A
best=B:110 elapsed≈0.30s

结合场景再看四个关注点

总耗时≈最慢上游，这是 I/O 并发的收益形态。
async def 调用不立即跑完——得到 coroutine，需被 await 或 task 调度。
time.sleep(0.3) 若误用，会阻塞整个循环，三个任务变串行。
CPU 密集比价算法本身不会因 async 变快；该换进程池或原生扩展。

核心概念与准确模型

1. 协程、任务、事件循环

概念	含义
协程函数	`async def f(): ...`
协程对象	`f()` 的返回值，需驱动
Task	被循环调度的并发单元（`create_task`）
事件循环	就绪队列 + I/O 多路复用 + 回调/任务调度
awaitable	可被 `await`：coroutine / Task / Future 等

flowchart LR
  Run["asyncio.run"] --> Loop[事件循环]
  Loop --> T1[Task 上游A]
  Loop --> T2[Task 上游B]
  Loop --> T3[Task 上游C]
  T1 -->|await I/O| Loop
  T2 -->|await I/O| Loop
  T3 -->|await I/O| Loop

2. 入口 `asyncio.run`

应用主入口推荐：

asyncio.run(main())

它创建循环、跑到完成、清理待定结构。在已有循环内（如 Jupyter、框架）不要嵌套 asyncio.run。

3. 并发组合

await asyncio.gather(c1, c2)：一起等；默认可收集异常（return_exceptions）
asyncio.create_task：先启动后台任务
3.11+ TaskGroup：结构化并发（失败取消同组）——细节见任务、超时与取消

4. 阻塞点分类

代码	对循环的影响
`await asyncio.sleep` / 真异步 socket	可切换
`time.sleep` / 同步 `requests.get` / 重 CPU	卡住循环
`await asyncio.to_thread(fn)`	把阻塞卸到线程池

5. 与线程/进程选型（回顾）

负载	优先
高并发 I/O、可异步库	asyncio
阻塞第三方 SDK	线程池 / `to_thread`
CPU 密集	多进程 / 原生扩展
简单脚本	先同步

见并发模型总览、GIL 与线程。

6. 异步上下文与迭代（预告）

async with：异步资源（连接池）
async for：异步迭代器（流式响应）

它们依赖 __aenter__ / __aiter__ 等协议，与同步 with/for 平行。

设计动机

用顺序代码表达并发控制流，比回调地狱可读。
在单线程内支撑大量等待连接，降低每连接线程成本。
显式 await 点让切换点可见——“哪里可能让出”写在代码里。

代价：

生态需要异步化，混入阻塞会静默伤吞吐
取消/超时/异常传播比同步更绕
调试栈与线程模型不同

边界情况与反直觉行为

1. 忘记 await

fetch_quote("A", 0.1, 1)  # 只创建 coroutine，通常警告且不执行业务

2. 在协程里 CPU 空转

长时间纯 Python 循环不 await，其他 Task 饿死。

3. 线程安全默认假设不成立

多数 asyncio 原语绑定当前循环；跨线程调度要用专用 API（call_soon_threadsafe 等）。

4. 取消是协作式

task.cancel() 在 await 点注入 CancelledError；必须在 finally 里释放资源。详见超时与取消专篇。

5. `gather` 与异常

默认一个子任务失败会在 await gather 时抛出；其他任务可能仍在跑——生产更倾向 TaskGroup 或显式取消策略。

常见误区

[!warning] 常见误区：async = 多核并行错误理解：写了 async 就能吃满 8 核。
正确模型：默认协作式并发；CPU 并行另选模型。

[!warning] 常见误区：把同步库直接丢进 async 路由错误理解：函数声明 async 就会变非阻塞。
正确模型：只有 await 真异步 I/O 或显式 to_thread 才让出。

[!warning] 常见误区：用 sleep 假装同步/锁错误理解：await asyncio.sleep 可替代正确同步。
正确模型：sleep 只是定时让出；共享状态用队列/锁。

[!warning] 常见误区：创建 Task 后不管错误理解：fire-and-forget 最省事。
正确模型：要有人 await/回收异常，否则静默失败与泄漏。

与相邻概念对比

模型	重叠等待	多核 CPU	典型成本
多线程	可	GIL 下 Python CPU 弱	共享内存同步
多进程	可	强	序列化/内存
asyncio	强（I/O）	弱（单循环线程）	阻塞敏感、取消复杂
同步	弱	弱	简单正确

文字说明：选型看瓶颈类型，而不是看哪套语法更新潮。

工程实践

服务内

Web 框架异步路由中禁止重 CPU 与同步 HTTP
统一超时：每个外部 await 都有预算
连接池用 async 驱动（DB/HTTP）

可观测

记录 in-flight 任务数、超时率、阻塞告警
关键路径带 correlation id

测试

asyncio.run 或 pytest-asyncio
用短 sleep 测并发时序，但不断言过紧的绝对时间

与 AI 应用

Agent 扇出多工具/多检索时，asyncio 很适合；但 tokenizer/本地推理 CPU 段要隔离。

可验证实验

实验 1：并发 vs 串行

把 gather 改成顺序三次 await，观察 elapsed 从 ~0.30s 变为 ~0.75s。

实验 2：误用 time.sleep

在 fetch_quote 里换成 time.sleep(0.2)（记得 import），观察总耗时接近相加且 start/done 日志不再交错。

实验 3：未 await

临时写 fetch_quote("A", 0.1, 1) 不 await，运行看 RuntimeWarning。

实验 4：to_thread（扩展）

用 await asyncio.to_thread(time.sleep, 0.2) 对比直接 time.sleep 对并发的影响。

本节总结

asyncio 的核心收益是 I/O 等待重叠，不是自动加速计算。
先掌握 async/await、run、gather，再进入 Task 生命周期与取消。
生产正确性 = 无阻塞热点 + 超时取消 + 异常回收。

自测题

概念题

协程函数与协程对象差别？
为什么说 asyncio 通常不解决 CPU 密集？

代码推理题

三个 await asyncio.sleep(0.2) 顺序执行与 gather 并发，耗时量级差多少？

工程思考题

异步 FastAPI 路由里调用同步 requests.get 会有什么风险？如何改？

参考答案

async def 定义函数；调用它返回 coroutine 对象，需被调度执行。
事件循环线程在跑 Python 计算时难以切换，且默认不跨多核并行解释。
顺序约 0.6s，并发约 0.2s（忽略调度噪声）。
阻塞 worker/事件循环，拖垮并发；改 httpx/aiohttp 异步客户端，或 to_thread/线程池隔离。

延伸阅读与资料来源

资料	类型	支撑内容
asyncio	文档	官方总览
Coroutines and Tasks	文档	Task/gather/timeout
PEP 492	PEP	async/await 语法
PEP 3156	PEP	异步 IO 循环愿景（历史）

笔记元信息

角色：原子概念（asyncio 主叙事）
深度：场景证据 + 模型 + 边界 + 实验
取消/超时专篇：python-asyncio-tasks-timeouts-and-cancellation

Python 异步编程

这个概念为什么出现

最小可运行示例

场景：价格聚合服务同时查询三个上游报价

结合场景再看四个关注点

核心概念与准确模型

1. 协程、任务、事件循环

2. 入口 asyncio.run

3. 并发组合

4. 阻塞点分类

5. 与线程/进程选型（回顾）

6. 异步上下文与迭代（预告）

设计动机

边界情况与反直觉行为

1. 忘记 await

2. 在协程里 CPU 空转

3. 线程安全默认假设不成立

4. 取消是协作式

5. gather 与异常

常见误区

与相邻概念对比

工程实践

服务内

可观测

测试

与 AI 应用

可验证实验

实验 1：并发 vs 串行

实验 2：误用 time.sleep

实验 3：未 await

实验 4：to_thread（扩展）

本节总结

自测题

概念题

代码推理题

工程思考题

延伸阅读与资料来源

笔记元信息

2. 入口 `asyncio.run`

5. `gather` 与异常