一个很典型的现场：服务要调 20 个下游接口，同步写法串行 20 次、每次等 50ms，接口耗时就堆到 1 秒。换成 asyncio 后耗时降到接近最慢那一次，但 CPU 并没有变快。异步解决的是"等待"，不是"计算"，分清这条线，才不会把它用在图像处理、加解密、批量计算上然后抱怨没提速。

---

1. 异步解决什么：并发等待，而非并行计算

场景	顺序执行	并发后（按等待时间估算，非实测）	原因
20 次 HTTP 调用（各 50ms）	20×50ms=1s	接近最慢的一次	等待期让出执行权
time.sleep(0.2) ×10	2s	仍为 2s（真阻塞）	阻塞调用不让出，退化为串行
纯计算 200ms ×10	2s	仍为 2s，甚至更慢	无等待可重叠，只剩调度开销
大文件 read() 阻塞读	阻塞整个循环	需 to_thread	磁盘 IO 在 asyncio 里不算"可等待"

结论：异步的收益 = 可等待时间 × 并发度。等待时间占比低、或者等待点被阻塞调用堵住，收益就是零。

---

2. 最小模型：async def / await / asyncio.run

import asyncio

async def fetch(uid: int) -> str:
    await asyncio.sleep(0.05)          # 模拟网络等待
    return f"user-{uid}"

async def main() -> None:
    r = await fetch(1)                 # 在协程内部必须 await
    print(r)                           # 典型输出：user-1

if __name__ == "__main__":
    asyncio.run(main())                # 程序入口只调用一次

写法	得到什么	说明
async def f()	协程函数	调用它不会执行函数体
f()	协程对象	未 await 会有 RuntimeWarning: never awaited
await f()	函数返回值	只能在协程内部用
asyncio.run(main())	事件循环跑完的最终结果	创建并关闭循环，一个进程只该调一次
asyncio.create_task(c)	Task	立刻排入循环，可稍后 await

---

3. 事件循环与协程调度

事件循环是一个单线程的调度器：维护就绪队列与等待中的 IO，await 让出控制权时把协程挂起，IO 就绪后重新入队。

概念	作用	实践含义
事件循环	单线程跑所有协程	一个慢操作会拖住全部任务
await 点	让出执行权的唯一位置	没有 await 的长循环会独占循环
Task	被循环调度的协程包装	需要"并发"就必须先变成 Task

关键点：await 只是"可能让出"，不保证并发。await a(); await b() 是串行的；要并发必须先把它们变成 Task 并交出控制权。

---

4. gather / wait / as_completed

# 全都要，任一失败就整体失败（默认）
results = await asyncio.gather(fetch(1), fetch(2), fetch(3))

# 全都要，但要拿到每个结果或异常
results = await asyncio.gather(*coros, return_exceptions=True)

# 等到满足条件就继续，不取消其余
done, pending = await asyncio.wait(tasks, timeout=1.0, return_when=asyncio.FIRST_COMPLETED)

# 按完成顺序处理，适合流式消费
for fut in asyncio.as_completed(tasks):
    print(await fut)

API	结果顺序	异常行为	适用
gather	与传入顺序一致	默认第一个异常向上抛，其余任务仍在跑（可能泄漏）	结果要按序对齐
gather(return_exceptions=True)	与传入顺序一致	异常作为返回值	批量任务要全量报告
wait	两个集合	不抛异常，异常留在 Task 上	需要超时/部分完成
as_completed	完成顺序	迭代时 await 才抛	边完成边处理

gather 最容易被忽略的行为：某个协程抛异常时，其他任务不会被自动取消，它们继续跑并可能在后台报 Task exception was never retrieved。要严格控制生命周期，用 TaskGroup。

---

5. TaskGroup（3.11+）与超时控制

import asyncio

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch(1))
        t2 = tg.create_task(fetch(2))
    print(t1.result(), t2.result())

# 超时：3.11+ 用 asyncio.timeout
async def with_timeout() -> str:
    async with asyncio.timeout(2.0):
        return await fetch(1)

特性	gather	TaskGroup
任一失败	其余继续跑	其余被取消，抛 ExceptionGroup
取消语义	需手工 cancel	结构化，退出 with 即收敛
异常类型	原始异常	ExceptionGroup，需 except* 拆
适用版本	3.x 全版本	3.11+

TaskGroup 的价值是结构化并发：块内任务不可能逃逸到块外还在跑。处理异常写 except* ValueError as eg: 逐个 eg.exceptions 取；3.11 之前等价物是 gather + 手工 cancel + try/finally。

---

6. 阻塞调用：一行 time.sleep 卡死整个循环

import time

async def bad() -> None:
    time.sleep(1)          # 灾难：整个事件循环停 1 秒
    requests.get(URL)      # 灾难：DNS + TCP + 读全在阻塞

async def good() -> None:
    await asyncio.to_thread(time.sleep, 1)          # 3.9+，最简写法
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, requests.get, URL)   # 复用默认线程池

阻塞源	是否让出循环	处理
time.sleep	否	asyncio.sleep
requests.*	否	httpx.AsyncClient 或 to_thread
同步 DB 驱动（pymysql/psycopg2）	否	线程池，或换 async 驱动
open().read() 大文件	否	asyncio.to_thread
subprocess.run	否	asyncio.create_subprocess_exec
CPU 密集计算	否	多进程，不是线程

判断标准很简单：这个函数内部有没有 await？ 没有，它就是阻塞的，必须扔进线程池或换异步实现。run_in_executor(None, ...) 用默认 ThreadPoolExecutor，并发上限受线程池大小限制，长时间任务要显式建池。

---

7. 异步 HTTP 客户端：httpx 与 aiohttp

import httpx

async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=5.0, http2=True) as client:
        tasks = [client.get(u) for u in urls]
        resps = await asyncio.gather(*tasks)
        return [r.json() for r in resps]

客户端	定位	注意
httpx.AsyncClient	与 requests 风格接近，同步/异步同 API	客户端要复用，别每次请求新建
aiohttp.ClientSession	老牌异步 HTTP	同理必须复用 session，且要在协程内创建
requests	纯同步	在协程里用等于阻塞循环

复用客户端的意义不只是性能：连接池、cookie、HTTP/2 复用都挂在 client 上，每请求新建一个 client 等于每次重新握手。

---

8. 限并发：Semaphore

sem = asyncio.Semaphore(10)

async def bounded(u: str) -> str:
    async with sem:
        return await fetch(u)

results = await asyncio.gather(*[bounded(u) for u in urls])

控制手段	粒度	适用
Semaphore(n)	同时运行的任务数	保护下游，最常用
asyncio.timeout(t)	单次耗时	防慢请求拖尾
分批 for chunk in chunks	批次	需要进度与失败重试
连接池 limits	连接数	httpx 的 Limits(max_connections=...)

Semaphore 控制的是同时进入临界区的协程数，不是每秒请求数。要做 QPS 限流得另加令牌桶；把它当限速器用，会在下游变慢时自动降低吞吐，看着像限流，实际是拥塞。

---

9. CPU 密集：该用多进程

import concurrent.futures as cf

def heavy(n: int) -> int:
    return sum(i * i for i in range(n))

with cf.ProcessPoolExecutor() as pool:
    out = await asyncio.get_running_loop().run_in_executor(pool, heavy, 10**6)

任务类型	正确工具	原因
IO 等待多	asyncio（单线程）	等待期可重叠
CPU 计算多	ProcessPoolExecutor	绕开 GIL，真并行
混合：IO + 少量 CPU	asyncio + to_thread	少量 CPU 不值得开进程
大量小任务	批量化后再并行	进程启动与序列化开销高

线程池对 CPU 密集无效——GIL 让同一时刻只有一个线程执行字节码。进程池的代价是序列化（pickle）与启动开销，单个任务耗时小于几十毫秒时，并行收益会被开销吃掉。

---

10. 排错

现象	根因	处理
RuntimeWarning: coroutine was never awaited	调了协程函数没 await	补 await 或 create_task
并发后耗时没变	内部用了同步阻塞调用	逐个函数检查有无 await 点
RuntimeError: asyncio.run() cannot be called from a running event loop	在协程/Jupyter 里再调 run	用 await main()，Jupyter 用顶层 await
Event loop is closed	循环关闭后仍有任务/客户端在用	客户端在 within 协程内创建并关闭
Task exception was never retrieved	Task 抛异常没人 await	用 TaskGroup 或收集后统一 await
Task got Future attached to a different loop	对象跨循环复用（如全局 client）	每个循环用独立客户端
请求全都串行	await 写在 for 循环里	先收集协程再 gather

倒数第一条是最常见的性能事故：for u in urls: await client.get(u) 语义正确但完全串行。改成先构造协程列表再 gather，才真正并发。

---

11. 边界

场景	更合适的选择	理由
脚本里跑几个 IO 任务	同步 + 线程池	异步的调试成本高于收益
需要严格背压与流控	asyncio + 队列/信号量	但要自己管，复杂度高
Web 框架已在异步栈上	用框架的异步视图	别混同步阻塞
大量短连接爬取	asyncio + 连接池	收益最明显
需要任务持久化与重试	任务队列（Celery/RQ）	内存里的 Task 不持久
单机数据管道	多进程 / DuckDB	与 IO 并发无关

asyncio 不是"更快的 Python"，而是"更适合等待的 Python"：把同步脚本机械地加 async/await，只会得到更难调试、同样慢的代码。

---

12. 自测

1. 20 次各 50ms 的 HTTP 调用，异步能压到多少？如果其中一次换成 time.sleep(0.2)，结果会怎样？
2. await a(); await b() 与 await asyncio.gather(a(), b()) 的执行顺序有何不同？
3. gather 中一个协程抛异常时，其余任务会怎样？TaskGroup 呢？
4. 怎么判断一个函数在协程里是不是"阻塞"的？发现阻塞后有哪两种处理方式？
5. Semaphore(10) 限制的是并发数还是 QPS？为什么它不能当限流器用？
6. CPU 密集任务放进 run_in_executor 的默认线程池，为什么没有加速？