把 requests.get 塞进 async def 里，并发度还是 1；把计算函数丢进 ThreadPoolExecutor，CPU 还是只用满一个核。这两件事的共同点是：选了并发模型，却没选对瓶颈类型。判断标准其实只有一句：时间花在等还是花在算。

---

1. 判定标准：先量，再选

特征	瓶颈	首选	不该选
时间在 recv/connect/read 上	IO 等待	asyncio 或线程池	多进程（白付启动成本）
CPU 占满单核，top 里 100%	计算	多进程 / C 扩展	asyncio（只会更慢）
每任务几毫秒、总量上百万	调度开销	批处理 + 向量化	一任务一协程
混合：取数后算	两段都要	异步取 + 进程池算	全塞进一个模型
依赖阻塞库且不可替换	IO 但无异步驱动	asyncio.to_thread	硬写 await 假异步

量化方式很朴素：time.perf_counter() 包住"等"与"算"两段，各记总时长。等待占比七成以上走异步/线程，计算占比七成以上走多进程。

---

2. asyncio 的模型与它不能做的事

事件循环是单线程的调度器：协程在 await 处让出控制权，循环去跑下一个就绪任务。协程不是并行，是交错——同一时刻只有一个协程在跑 Python 字节码。

概念	实际含义	常见误解
事件循环	单线程任务调度器	以为什么都自动并行
await	让出控制权的挂起点	以为是"开新线程"
asyncio.sleep(0)	主动让出一次调度	以为能提速
阻塞调用	卡住整个循环	以为只卡住当前协程
CPU 密集协程	独占循环，其他任务饿死	以为多开协程就能压满多核

import asyncio, time

async def tick(name: str, n: int) -> None:
    for i in range(n):
        print(f"{name}-{i}")
        await asyncio.sleep(0)      # 让出控制权，别的协程才有机会跑

async def main() -> None:
    await asyncio.gather(tick("a", 3), tick("b", 3))   # 交错输出 a-0 b-0 a-1 ...
    t0 = time.perf_counter()
    await asyncio.sleep(0.05)
    print("sleep 0.05 ->", round(time.perf_counter() - t0, 3), "s")

asyncio.run(main())

把上面任意一个 await asyncio.sleep(0) 换成 time.sleep(0.01)，输出立刻变成串行——这就是"阻塞调用卡住整个循环"的最小复现。

---

3. 阻塞库的三种处理

手段	写法	适用	代价
换异步驱动	httpx.AsyncClient、asyncpg、aiofiles	长期主力路径	要改调用点与依赖
丢线程	await asyncio.to_thread(fn, *args)	偶发阻塞、第三方 SDK	受默认线程池上限约束
显式执行器	loop.run_in_executor(pool, fn)	需要自管线程池大小	代码更啰嗦

import asyncio, httpx

async def fetch_all(urls: list[str]) -> list[int]:
    async with httpx.AsyncClient(timeout=10) as client:      # 换异步驱动
        async def one(u: str) -> int:
            r = await client.get(u)
            return r.status_code
        return await asyncio.gather(*(one(u) for u in urls))

def legacy_blocking(path: str) -> bytes:
    with open(path, "rb") as f:
        return f.read()

async def read_many(paths: list[str]) -> list[bytes]:
    return await asyncio.gather(*(asyncio.to_thread(legacy_blocking, p) for p in paths))

to_thread 的默认池上限与 min(32, os.cpu_count() + 4) 相关，它不会因为协程开了一万个就给你一万个线程；要更高并发得自己建 ThreadPoolExecutor 并 run_in_executor。

---

4. multiprocessing 的开销与进程池

开销项	说明	规避
进程启动	每个 worker 重新 import 主模块	用 if __name__ == "__main__": 保护
参数序列化	每次 submit 都 pickle	传小对象；大数组用共享内存
返回值序列化	结果也要 pickle 回来	让 worker 自己落盘，只回路径
不可序列化对象	锁、连接、lambda、局部函数	在 worker 内部创建
内存不共享	每个进程独立堆	只读数据走 fork（Linux）或 mmap

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count

def cpu_heavy(n: int) -> int:      # 必须是模块级函数，才能被 pickle
    total = 0
    for i in range(n):
        total += i * i
    return total

def main() -> None:
    with ProcessPoolExecutor(max_workers=cpu_count()) as pool:
        print(sum(pool.map(cpu_heavy, [200_000] * 8)))

if __name__ == "__main__":         # spawn 平台必需
    main()

数值型大数组别用 ProcessPoolExecutor 硬传：一次 pickle 的拷贝成本往往超过计算本身，multiprocessing.shared_memory 或直接把分片文件名传给 worker 更实在。

---

5. GIL 对线程与 CPU 密集的真实影响

任务类型	线程有效吗	原因
网络 IO	有效	等待时释放 GIL
磁盘 IO	部分有效	页缓存命中时几乎不等待
纯 Python 计算	几乎无效	字节码执行需持有 GIL
numpy/zlib/hashlib 大块运算	有效	C 扩展内部释放 GIL
调用外部进程	有效	等子进程时释放

CPython 3.11 仍有 GIL（可选子解释器是后续路线）。结论很实用：线程池在 IO 上依然值钱，别因为"有 GIL"就把线程方案一刀切掉。

---

6. 并发控制的三个旋钮

旋钮	异步写法	进程/线程写法
并发上限	asyncio.Semaphore(n)	max_workers=n
超时	asyncio.timeout(s)（3.11+）	future.result(timeout=s)
重试	装饰器 + asyncio.sleep 退避	同左（注意别在 worker 内重试）

import asyncio, random

async def fetch(i: int) -> str:
    await asyncio.sleep(random.uniform(0.01, 0.05))
    if i % 7 == 0:
        raise RuntimeError(f"bad {i}")
    return f"ok-{i}"

async def bounded(ids: list[int], limit: int = 8) -> list[str]:
    sem = asyncio.Semaphore(limit)
    async def one(i: int) -> str:
        async with sem:
            async with asyncio.timeout(2):
                return await fetch(i)
    results = await asyncio.gather(*(one(i) for i in ids), return_exceptions=True)
    return [r for r in results if isinstance(r, str)]

print(len(asyncio.run(bounded(list(range(30))))))

Semaphore 要在创建任务之前建好并共享，别在 one() 里新建——那样每个任务各有自己的信号量，等于没限制。

---

7. 背压与结果收集

方式	异常行为	适用
gather(...)	首个异常向上抛，其余任务不被取消且结果丢弃	全成功才继续的场景
gather(..., return_exceptions=True)	异常作为值返回	需要逐项判定
asyncio.TaskGroup()（3.11+）	任一失败即取消同组其余任务，抛 ExceptionGroup	强一致的一组任务
asyncio.as_completed	按完成顺序产出	边完成边处理
asyncio.Queue(maxsize=n)	队列满时 await put 阻塞生产者	生产者远快于消费者

import asyncio

async def guarded(items: list[int]) -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            for i in items:
                tg.create_task(work(i))
    except* RuntimeError as eg:        # 3.11 的 except* 语法
        print("failures:", len(eg.exceptions))

async def work(i: int) -> None:
    await asyncio.sleep(0)
    if i == 2:
        raise RuntimeError("boom")

gather 在第一个异常处返回，但已在跑的任务不会被取消，它们的异常会以 "Task exception was never retrieved" 的形式出现在日志里。用 TaskGroup 能得到确定的取消语义，代价是必须处理 ExceptionGroup（except*）。

---

8. 子进程：别在协程里 subprocess.run

import asyncio

async def run_cmd(cmd: list[str]) -> tuple[int, bytes]:
    proc = await asyncio.create_subprocess_exec(
        *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
    out, _ = await proc.communicate()
    return proc.returncode, out

print(asyncio.run(run_cmd(["python", "-c", "print('hi')"])))

注意点	说明
用 subprocess.run 在协程里	阻塞循环，等于串行
不读 stdout 就 wait()	管道缓冲区满，子进程卡死
不设超时	子进程挂住，协程永远不返回
shell 拼接字符串	参数注入；用列表形式传参
忘了 communicate()	僵尸进程 + 未读输出

---

9. 排错：六种典型现场

现象	根因	处理
并发度上不去	协程里调了 requests/time.sleep	换异步驱动或 to_thread
只收到部分结果	gather 首个异常导致其余丢弃	return_exceptions=True 或 TaskGroup
进程池报 PicklingError	传了 lambda、连接、局部函数	改模块级函数，连接在 worker 内建
日志提示协程从未执行	忘了 await，只创建了协程对象	加 await 或 create_task
程序卡在退出	后台任务未取消、执行器没 shutdown	async with / 显式 cancel()
多进程后内存翻倍	每个 worker 各持一份只读数据	共享内存或 fork 复用

---

10. 边界：这些情况本就不该用并发

需求	更合适的选择	原因
单次请求延迟已达标	保持同步代码	并发只增加复杂度
需要严格顺序的写操作	单线程队列	并发下顺序不可控
计算本身在 C 库内已并行	直接调库（numpy、torch）	再套进程会超额订阅
任务数远大于 CPU 且每个极短	批处理 / 向量化	调度开销吃掉收益
需要跨请求共享可变状态	数据库 / 缓存	进程间共享内存极易出错

一句话：并发是放大器，放大的是吞吐，也放大错误处理与状态的复杂度。 不确定瓶颈在哪时，先加计时，再选模型。

---

11. 自测

1. 判断"该用异步还是多进程"时，你最先测的是哪两个量？
2. 为什么在协程里调用 requests.get 会让并发度退化成 1？
3. asyncio.to_thread 的并发上限受什么约束？
4. gather 抛异常时，其他已在运行的任务会怎样？
5. TaskGroup 与 gather 在失败语义上的关键差别是什么？
6. 进程池里传一个数据库连接对象，会在哪一步失败？