第25篇|asyncio 异步编程
本篇定位:《Python 教程》专栏二「Python 进阶与工程」第 25 篇。承接第 23~24 篇把「线程叠 I/O、进程扛 CPU」说清之后,正面讲清 asyncio:单线程事件循环如何叠成千上万个等待、
async/await如何协作切换、以及何时该把阻塞活丢给线程池 / 进程池。为第 26 篇 HTTP(httpx异步客户端)、第 27 篇 FastAPI、第 28 篇 Agent 工具循环打底。读完你将能够:用自己的话解释「异步为何能叠等待」;写出
async def/await/asyncio.run的最小可运行程序;用create_task、gather、as_completed编排并发;识别「在环里跑time.sleep/ 同步requests」的堵死;判断 AI Infra 场景里协程、线程池、进程池该各放哪一层。预计用时:阅读 35~45 分钟,动手 45~60 分钟。
第 23 篇给了同进程线程:适合叠 I/O,但每个线程有栈与调度成本。第 24 篇给了跨进程并行:适合纯 Python CPU,但启动与序列化有税。本篇回答第三个工程问题——「连接数上千、等待时间很长时,单线程怎么高效叠?」
一句话定心:asyncio 不加速算力,而是把「等网络 / 等磁盘 / 等定时器」的空隙叠在一起;代价是协作式让出——你必须在可等待点 await,中间跑阻塞代码会卡住整条环。 线程靠操作系统抢占;协程靠你主动让出。
建议边读边在练习目录里动手(uv init --no-package,标准库 asyncio 即可;第 26 篇再装 httpx)。阅读顺序:心智模型 → async/await → 事件循环与 Task → 编排 API → 阻塞与 Executor → AI Infra → 踩坑。下一篇预告网络请求(26):同步 requests 与异步 httpx、如何并发打大模型 API。
和第 15 篇对照:协程函数返回的是「可等待对象」,有点像「懒执行的生成器协议」——但你要用 await 驱动,而不是 next。和第 23 篇对照:同一段「等 1 秒 × 100」,线程池开 100 个线程很重,asyncio.gather 开 100 个 Task 很轻。和第 24 篇对照:CPU 重循环仍别指望 asyncio——请 run_in_executor 丢进程池。
动手时建议沿用「并发打多个假 API」意象:主协程编排;每个 Task await asyncio.sleep 模拟网络;gather 收集结果。真正的 HTTP 客户端留给第 26 篇。做完后打开任意 FastAPI 路由或 Agent 并行工具调用,你会认出同一套形状。
本篇不穷尽所有底层协议(Protocol / 传输层)、不替代完整异步框架教程(Trio / anyio 细节)、不深入 CPython 事件循环 C 实现。主线是:等待可叠 → 语法与环 → Task 编排 → 阻塞边界 → 选型与 AI 场景 → 踩坑。和「听说异步就全改 async」相比,正确价值是:在 I/O 边界叠协程,在阻塞边界挪出去,在编排边界保持清晰。
写异步时保持一个工程直觉:能 await 的才进环;不能 await 的用 Executor;限流用 Semaphore;取消要会处理。 「在 async def 里直接 time.sleep」会让整服务假死——三个月后的夜班你会感谢今天的肌肉记忆。
也可以把本篇看成专栏二「并发面」的收官:第 23 篇同进程叠任务,第 24 篇跨进程真并行,本篇单线程叠海量 I/O。后面 HTTP、FastAPI、Agent,都假设「你知道何时开协程、何时开池」。
读官方 asyncio 文档时,按本篇目录对照即可。入门阶段把「会画事件环、会写 gather、会避开阻塞、会选型」做扎实,比背完整 API 表更重要。
动手时再记一条:异步的第一原则仍是正确优先于炫——错误的 async 包装比同步更糟,因为它摧毁可读性并引入取消 / 竞态新坑。排障时一条最小可复现(两个 sleep Task 对照),往往比一百次「再加并发」更值钱。
和「一切皆异步」的神话相比,更值钱的是匹配瓶颈:等的是网络 → 协程;等的是 CPU → 进程;少量阻塞库暂时甩不掉 → 线程池。把这个品味带进 code review,比强制「必须 asyncio」温和且有效。
若你刚从第 24 篇「进程与 pickle」过来,本篇前两节会换一种体感:同样是叠任务,但挂钩点从「隔离与序列化」换成「让出与就绪队列」。把「串行三次 sleep(1)」改成 gather 三次,把「环里 time.sleep」改成 asyncio.sleep——两条路径走完,异步直觉就长出来了。
和「在聊天里口头说异步」相比,可运行示例的价值是:机器可重复、可测、可对比墙钟时间——本地与 CI 上同一套结论。排障时你面对的是「谁堵死了环」,而不是「我记得好像快了一点」。
动手前再立一条纪律:先同步正确,再异步。 先写串行版本并测绿,再引入 async/await,最后用计时验证是否真有收益。很多人倒过来「先全改 async 再修阻塞」,修的时候容易不自觉把同步客户端塞进协程——事件环变成了复杂度放大器。小步红绿循环在异步里同样成立。
一、为什么要谈 asyncio:把等待叠起来
1.1 一张图看清同步与异步

图1 同步串行 vs 异步并发:同一段墙上时间
import asyncio
import time
async def fake_io(name: str, delay: float) -> str:
await asyncio.sleep(delay) # 模拟网络等待
return f"{name}-ok"
async def main() -> None:
t0 = time.perf_counter()
# 串行:三次等待相加
a = await fake_io("A", 1.0)
b = await fake_io("B", 1.0)
c = await fake_io("C", 1.0)
print("serial", a, b, c, f"{time.perf_counter() - t0:.2f}s")
t0 = time.perf_counter()
# 并发:墙钟 ≈ 最慢那一次
results = await asyncio.gather(
fake_io("A", 1.0),
fake_io("B", 1.0),
fake_io("C", 1.0),
)
print("gather", results, f"{time.perf_counter() - t0:.2f}s")
if __name__ == "__main__":
asyncio.run(main())
serial A-ok B-ok C-ok 3.00s
gather ['A-ok', 'B-ok', 'C-ok'] 1.00s
(秒数因机器略有抖动;趋势稳定:串行约 3s,并发约 1s。)这就是 asyncio 存在的理由——不是把 1+1 算得更快,而是把三次「等 1 秒」叠成一次墙钟。
要点:
- 异步 = 重叠等待:事件循环在某个 Task 等待时,去跑别的就绪 Task。
- 单线程 = 无数据竞争于字节码切换:但仍可能有「共享可变逻辑竞态」(见踩坑)。
- 协作 = 必须
await:没有 await 的长时间计算,环转不动。
1.2 和第 23、24 篇的边界
第 23 篇:线程适合中等 I/O,但一万个连接时线程栈与上下文切换吃紧。第 24 篇:进程适合 CPU,对「等 HTTP」往往过重。本篇:海量可等待 I/O 的默认选型。三者互补——混合流水线很常见:协程管并发请求,进程池管重解析,GPU 服务独占推理。
AI Infra 视角:并发打几十上百个 LLM / 工具 API、流式读 token、FastAPI 高并发路由——典型 asyncio 场景;离线扫百万条做特征——仍偏进程池。先问「等的是网络还是算力」,再问「调用链是否全程可 await」。
1.3 渐进采用
老脚本全是同步 for + time.sleep / 同步 HTTP?可以:先把「一次 I/O」提成 async def;用 asyncio.run 包入口;用 gather 替换串行循环。不必一天重写整站,但新的海量 I/O 边界不要再无脑开几千线程。
二、async / await:协程是什么
2.1 协程函数与协程对象
import asyncio
async def add(x: int, y: int) -> int:
return x + y
coro = add(1, 2) # 还没执行!只是协程对象
print(type(coro))
result = asyncio.run(add(1, 2)) # 真正跑起来
print(result)
<class 'coroutine'>
3
关键心智:调用 async def 不会立刻跑函数体,只得到协程对象;必须交给事件循环(asyncio.run / await / create_task)才会执行。这和第 15 篇「调用生成器函数得到生成器对象」同构。
2.2 await:让出并等待
await 做两件事:(1)把当前协程挂起,把控制权交还事件循环;(2)等右侧「可等待对象」完成后,拿结果继续往下跑。
import asyncio
async def step() -> str:
print("before")
await asyncio.sleep(0.1)
print("after")
return "done"
asyncio.run(step())
before
after
可等待对象常见三类:协程、Task、实现了异步协议的对象(如部分库的 response)。对初学者:先记住「await 右边必须是 async 世界认可的东西」——await time.sleep(1) 会直接报错。
2.3 事件循环一张图

图2 事件循环:就绪 · 等待 · 回调续跑
asyncio.run(main()) 是应用入口的标准写法:创建循环 → 跑到完成 → 清理。在已经运行的循环里不要再 asyncio.run(会报错);嵌套场景用 await 或 create_task。
import asyncio
async def main() -> None:
print("hello async")
if __name__ == "__main__":
asyncio.run(main())
hello async
库代码一般不要自己 asyncio.run——那是应用入口的职责;库只提供 async def,让调用方决定如何驱动。
三、Task 与编排:create_task / gather / as_completed
3.1 一张图看并发编排

图3 并发编排:create_task · gather · as_completed
3.2 create_task:立刻调度
import asyncio
async def work(n: int) -> int:
await asyncio.sleep(0.05)
return n * n
async def main() -> None:
t1 = asyncio.create_task(work(3))
t2 = asyncio.create_task(work(4))
# 两个 Task 已在跑;下面 await 取结果
print(await t1, await t2)
asyncio.run(main())
9 16
create_task 把协程包装成 Task 并安排到循环上;即使你暂时不 await,它也会推进(直到下一个 await 点)。忘记保存 / await Task,可能被垃圾回收并发出警告——工程上:创建的 Task 要有主人。
3.3 gather:等全部(默认可保序)
import asyncio
async def fetch(tag: str, delay: float) -> str:
await asyncio.sleep(delay)
return tag
async def main() -> None:
results = await asyncio.gather(
fetch("A", 0.3),
fetch("B", 0.1),
fetch("C", 0.2),
)
print(results) # 按传入顺序,不是完成顺序
asyncio.run(main())
['A', 'B', 'C']
gather 适合「一批同类任务,要全部结果」。默认一个子任务异常会传染到 gather(可设 return_exceptions=True 变成结果里的异常对象)。
3.4 as_completed:谁先完谁先处理
import asyncio
async def fetch(tag: str, delay: float) -> str:
await asyncio.sleep(delay)
return f"{tag}:{delay}"
async def main() -> None:
tasks = [
fetch("A", 0.3),
fetch("B", 0.1),
fetch("C", 0.2),
]
for coro in asyncio.as_completed(tasks):
print(await coro)
asyncio.run(main())
B:0.1
C:0.2
A:0.3
(完成顺序可能因调度略有差异,但通常快的先打印。)需要「流式处理先到结果」用 as_completed;需要「整齐列表」用 gather。
3.5 Semaphore:限流
并发打 API 时,瞬时一万个连接会把自己或对方打挂。用信号量限制「同时进行中的 Task 数」:
import asyncio
async def hit(sem: asyncio.Semaphore, i: int) -> int:
async with sem:
await asyncio.sleep(0.05)
return i
async def main() -> None:
sem = asyncio.Semaphore(3) # 最多 3 个同时跑
results = await asyncio.gather(*(hit(sem, i) for i in range(10)))
print(results)
asyncio.run(main())
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
限流是生产异步的标配:和第 23 篇「有界队列」、第 24 篇「有界进程池」同向——并发要有上限。
四、阻塞陷阱与 run_in_executor
4.1 一张图看懂「堵死整环」

time.sleep(3)和同步requests.get,以及async def bad()函数,导致整环卡住3秒,其他Task全部饿死。正确做法是使用asyncio.sleep、异步库或线程池,如async def good()函数,通过await asyncio.sleep(3)让出控制权,环继续转,CPU重活→run_in_executor。图片与上下文紧密相关,直观呈现了在事件循环中跑阻塞代码的错误与正确处理方式。图4 危险:在事件循环里跑阻塞代码
import asyncio
import time
async def bad() -> None:
time.sleep(0.2) # 阻塞:环在这 0.2s 内转不动
print("bad done")
async def good() -> None:
await asyncio.sleep(0.2) # 让出:别人可以跑
print("good done")
async def main() -> None:
t0 = time.perf_counter()
await asyncio.gather(bad(), bad(), bad())
print("bad gather", f"{time.perf_counter() - t0:.2f}s")
t0 = time.perf_counter()
await asyncio.gather(good(), good(), good())
print("good gather", f"{time.perf_counter() - t0:.2f}s")
asyncio.run(main())
bad done
bad done
bad done
bad gather 0.60s
good done
good done
good done
good gather 0.20s
三次「坏 sleep」接近串行相加;三次「好 sleep」重叠。把 time.sleep 换成同步 requests.get,症状相同——服务里所有协程一起卡。
4.2 阻塞库怎么办:线程池
一时半会儿改不掉的同步函数,可以丢给默认线程池:
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
def blocking_io(n: int) -> int:
time.sleep(0.1) # 假装同步 SDK
return n
async def main() -> None:
loop = asyncio.get_running_loop()
with ThreadPoolExecutor(max_workers=4) as pool:
results = await asyncio.gather(
*[loop.run_in_executor(pool, blocking_io, i) for i in range(4)]
)
print(results)
asyncio.run(main())
[0, 1, 2, 3]
CPU 重活则换 ProcessPoolExecutor(第 24 篇)——同样经 run_in_executor 桥接,避免在环里跑纯 Python 死循环。
4.3 对照速查
五、取消、超时与结构化并发入门
5.1 wait_for:超时
import asyncio
async def slow() -> str:
await asyncio.sleep(2.0)
return "late"
async def main() -> None:
try:
await asyncio.wait_for(slow(), timeout=0.2)
except TimeoutError:
print("timeout")
asyncio.run(main())
timeout
(Python 3.11+ 内置 TimeoutError;旧文档常写 asyncio.TimeoutError,现为别名。)生产调用外部 API 必须设超时——无限等是事故温床。
5.2 取消:CancelledError
import asyncio
async def worker() -> None:
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
print("cleaned up")
raise # 通常继续抛出,让取消完成
async def main() -> None:
t = asyncio.create_task(worker())
await asyncio.sleep(0.05)
t.cancel()
try:
await t
except asyncio.CancelledError:
print("task cancelled")
asyncio.run(main())
cleaned up
task cancelled
取消是协作的:只有在 await 点才能被取消。在 finally 里释放资源;需要屏蔽取消的窗口用 asyncio.shield(高级,慎用)。
5.3 TaskGroup(3.11+)一瞥
import asyncio
async def one(n: int) -> int:
await asyncio.sleep(0.05)
return n
async def main() -> None:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(one(1))
t2 = tg.create_task(one(2))
print(t1.result(), t2.result())
asyncio.run(main())
1 2
TaskGroup 提供更结构化的「同生共死」:子任务异常会取消组内兄弟并向上抛出。新代码优先考虑它;本系列示例仍以 gather 为主,兼容面更广。
六、AI Infra:协程放哪一层
6.1 选型地图

图5 AI Infra:asyncio 适合放在哪一层
6.2 Agent 与工具并行
Agent 循环里「并行调用两个互不依赖的工具 HTTP」适合 gather:输入是 URL / JSON,输出是文本,状态机仍在主协程。可变的「当前对话状态」不要被多个 Task 同时瞎改——和第 22~24 篇同向:编排单点,工人尽量无共享可变。
6.3 和后续篇章的分工
- 第 23 篇:线程、GIL、锁、队列。
- 第 24 篇:多进程、
ProcessPoolExecutor、pickle。 - 本篇:事件循环、
async/await、Task 编排、阻塞边界。 - 第 26 篇:
requests/httpx、调用大模型 API(同步与异步对照)。 - 第 27~28 篇:FastAPI 与最小 Agent——默认在请求路径里写真 async,而不是假 async 包同步。
读到这里,不妨停一秒自检:你能不能画出「就绪队列 / 等待中 / 事件环」?能不能解释为何 time.sleep 会让 gather 退化成串行?能不能对「并发打 API」与「纯 Python 累加」分别给出建议?两问卡壳,就回到图 1、图 2、图 4 把箭头再走一遍。
和「等线上偶发再改超时」相比,现在付出的代价极低:wait_for、有界 Semaphore、禁止环内阻塞、入口 asyncio.run。事故夜里你会感谢今天的自己。也可以把本篇练习当成以后并发评测 / Agent 工具调用的骨架——限流与超时先约定好,比事后补齐便宜得多。
再强调一次边界:本篇教你在 CPython asyncio 默认模型下正确叠等待;anyio / Trio 结构化并发、uvicorn worker 模型、分布式队列是下一层,会在服务与 Infra 实战里逐步露面。不要因为「听说有更炫的框架」就今天在业务里省略超时与限流。
若你带团队做 code review,可以把本篇收成三条合并门槛:新增 async 是否真在等 I/O;环内是否出现同步阻塞;并发是否有 Semaphore / 超时。门槛不在于「必须全异步」,而在于不许用假异步换虚假吞吐。
七、踩坑大全
通用
坑 1:调用了 async def 却忘了 await / create_task。 现象:协程「从没跑」、RuntimeWarning。原因:只创建了协程对象。解决:await、create_task 或 asyncio.run。
坑 2:在协程里用 time.sleep 或同步 HTTP。 现象:整服务卡住、延迟尖刺。原因:阻塞事件循环。解决:asyncio.sleep / 异步客户端 / run_in_executor。
坑 3:在已有事件循环里再 asyncio.run。 现象:RuntimeError: ... running event loop。原因:嵌套 run。解决:在 async 上下文用 await;入口只 run 一次。
坑 4:并发无上限。 现象:文件描述符耗尽、对方 429、本机内存涨。解决:Semaphore;连接池限额;退避重试(第 26 篇展开)。
坑 5:gather 一个失败全灭,却没日志。 现象:难以排查。解决:return_exceptions=True 或包一层 try;统一记录。
坑 6:创建 Task 后既不 await 也不保存。 现象:任务被回收、警告「Task was destroyed」。解决:列表收纳;或用 TaskGroup。
坑 7:以为 asyncio 能加速纯 Python CPU。 现象:墙钟不降、还更卡。原因:单线程协作。解决:进程池(第 24 篇)。
macOS / Linux / Windows
坑 8:Windows 上部分策略与子进程配合怪异。 现象:ProactorEventLoop 与某些库不兼容。解决:查库文档;必要时显式设策略;优先用高层次 API。
坑 9:Linux 打开文件数默认偏低。 现象:高并发 OSError: Too many open files。解决:调 ulimit;降并发;复用连接。
坑 10:在 Jupyter / 交互式环境直接 asyncio.run。 现象:循环已存在报错。解决:await main()(notebook 已有环);或用 nest_asyncio(权宜,知代价)。
与 AI / 服务叠加
坑 11:FastAPI 路由声明了 async def 却内部狂调同步 requests。 现象:worker 假死。解决:换异步客户端;或改普通 def 让框架丢线程池(知代价)。
坑 12:流式生成里阻塞解析大 JSON。 现象:首 token 延迟偶发飙高。解决:解析挪出环;或分块增量处理。
八、小结
合上文章后,试着用自己的话说明:异步为何能叠等待、事件循环做什么、以及为何环内阻塞是大忌。讲得清,才算真的会了。动手至少跑通:一次串行 vs gather 对照 → 一次 create_task → 一次故意 time.sleep 看墙钟退化 → 一次 Semaphore 限流 → 一次 wait_for 超时。
- 叠等待:asyncio 把 I/O 空隙重叠;不加速纯算力。
- 语法:
async def产生协程;await让出并取结果;asyncio.run管入口。 - 编排:
create_task/gather/as_completed;生产必限流。 - 阻塞边界:环内禁
time.sleep/ 同步 HTTP;用异步库或 Executor。 - 超时与取消:
wait_for;正确处理CancelledError。 - AI Infra:HTTP / LLM / FastAPI 偏协程;CPU 偏进程;同步 SDK 用线程桥;编排单点。
把故事连起来:第 15 篇生成器协议 → 第 23 篇线程 → 第 24 篇进程 → 本篇单线程叠海量 I/O。专栏二工程刀具继续成套:下一站是网络请求与 HTTP(26)。
若只能带走一句话:等 I/O 用协程;阻塞请挪走;并发必限流加超时。
也可以用「三个不要」自检:不要忘了 await;不要在环里跑阻塞;不要无界海量 Task。
再留一个自测:把三次 sleep(1) 改成 gather 对照;故意用 time.sleep 看串行化;用 Semaphore(2) 跑 10 个任务——三问都顺,本篇就站稳了。
本篇技能会立刻复利:读开源先看它如何限并发与超时;PR 里要求说清是否真 async;重构争议先测量墙钟再加大并发。后续 HTTP、FastAPI、Agent,都假设这层已稳。
和专栏一串起来复习也不亏:生成器(15)帮你理解「惰性与让出」;异常(11)要在 gather / Task 里浮出;上下文管理(17)对应 async with;本篇是专栏二把「异步 I/O」说透的一课。
也可以把收获压成检查清单:会画事件环;会 asyncio.run;会 gather/as_completed;会避阻塞;会限流超时;能区分线程 / 进程 / 协程选型。六项过关,即可进入 HTTP 篇。
若把本篇压成给同事的三十秒讲解:先画「等待重叠」;再强调「await 才切换」;最后提「环内阻塞 = 全员挨冻」。三张草图比贴一篇博客更管用。等对方能独立把串行假 I/O 改成有界 gather,并解释为何 time.sleep 有毒,就可以一起 code review 异步 PR 了。
动手清单再压缩一版:写 fake_io 串行 vs gather;create_task 两枚;故意 time.sleep;Semaphore + wait_for。四步做完,比只「看完文章」扎实一个数量级。
九、课后练习
- (基础) 写
async def hello()打印一行,用asyncio.run启动;再故意只调用不 await,观察警告。 - (基础) 三个
asyncio.sleep延迟分别为 0.3 / 0.1 / 0.2,分别用串行 await 与gather对照墙钟。 - (基础) 用
create_task启动两个协程,在 main 里依次 await,打印返回值。 - (理解) 把
gather里的asyncio.sleep改成time.sleep,记录墙钟变化,用三句话解释原因。 - (理解) 用
asyncio.as_completed处理 5 个耗时不等的任务,打印完成顺序是否等于提交顺序。 - (理解) 用
Semaphore(2)包 8 个假请求,说明为何生产环境需要限流。 - (进阶) 组合
wait_for:让一个 1 秒任务在 0.2 秒超时,捕获异常并打印;再写一个刚好不超时的对照。 - (进阶 · AI Infra) 画一张你当前项目的并发图:哪些步骤用协程、哪些用线程池、哪些用进程池;写 200 字说明「假 async(async def 里调同步 HTTP)」的危害与改法。
下一篇预告
下一篇是《第26篇|网络请求与 HTTP:requests/httpx、调用大模型 API》——在搞清「协程如何叠等待」之后,解决「真实世界怎么发 HTTP、怎么并发打推理接口」。你会看到:同步 requests 的简单路径;httpx 的同步 / 异步双模;超时、重试与流式响应;以及调用大模型 API 时如何把本篇的 gather + Semaphore 落到实处。
专栏二后续路线:HTTP → FastAPI → 最小 Agent。带着本篇的事件环纪律与阻塞边界,写请求时记得:别在 async 路由里塞同步客户端;可等待的 I/O 才配 await;用测量与限流说话。
若想提前热身:把本篇 fake_io + gather + Semaphore 示例保存下来,下篇把 sleep 换成真 HTTP,对照线程池与协程在连接数上升时的资源差异。做完后第 26 篇谈 httpx.AsyncClient 时你会带着具体体感。
也可以把第 15~25 篇对照:生成器管流、装饰器管横切、上下文管资源、类型管形状、标准库给积木、依赖钉环境、日志给眼睛、测试给安全带、线程给同进程并发、进程给真并行、本篇给异步 I/O——工程刀具的轮廓已经清晰,下一篇把「网络请求」说透。
专栏一用十四篇把「对象与协议」立住;专栏二从迭代器(15)走到 asyncio(25),下一站是 HTTP 与大模型 API。下一篇见。
若时间充裕,再多做一件事:打开你最近用过的一个 Agent 或评测脚本,找到并发打 API 的代码,记三条「它如何限流、如何设超时、是否在 async 里调用了同步客户端」。这种阅读比空背 API 名更能把选型刻进肌肉记忆。
进入专栏二至此,可验证面、可观测面与并发面开始交汇:会看见问题,会自动验回归,也会在正确层级叠任务。下一篇开始把真实 HTTP 配齐。若某一节读着抽象,回到对应插图把箭头走一遍,再手打 gather 对照与 time.sleep 堵环——多数卡住的地方,其实是「线程、进程、协程」被混成同一句话。
写异步或读延迟报告时再记一条:错误往往来自错误层级与环内阻塞(该用协程却用忙等、该 await 却 time.sleep)。和「栈很长但根因很浅」相比,先选对并发模型能省很多时间。
也可以把本篇看成专栏二的「异步 I/O」补课:前面把协议、积木、环境、眼睛、断言与线程 / 进程收齐,本篇把协程收齐。后面 HTTP 篇教你如何把等待变成真实网络;FastAPI 与 Agent 篇则把异步打进真实请求与工具循环。
若阅读时觉得「线程、进程、协程」总搅在一起,回到图 1 与图 5:先定瓶颈类型,再选层。日常编程三者都要懂一点,但职责不同——混谈是多数初学者卡住的根源。
收官前对照练习:同一段「三个各等 1 秒的假 I/O」,分别用串行 for、线程池、asyncio.gather。你会清楚感受到:协程不是万能加速器,而是用协作换海量等待重叠的工具——也是专栏二工程化与 FastAPI / Agent 对接的入场券之一。
写完练习后,不妨在自己的仓库里开一个小 PR:只把串行假 I/O 循环改成有界 gather + Semaphore,不改业务语义。这种「小步叠异步」比空读文档更能把工作流刻进肌肉记忆,也更适合团队推广。
若你带新人,可以把本篇插图打印成一页速查:左列场景(HTTP / 纯循环 / FastAPI / Agent 编排),右列策略(协程 / 进程池 / 原生 async / 主协程)。新人写并发时对着贴,两周后就会自然说出「先量瓶颈」与「先禁止环内阻塞」。工具文化是带出来的,不是靠一次分享会喊出来的。
最后把专栏二的位置再钉一次:协议篇(15~18)解决「怎么正确接线」;标准库与依赖篇(19~20)解决「积木与环境」;可观测与可验证篇(21~22)解决「如何被看见与自动验证」;并发篇(23~25)解决「线程、进程、协程如何分层叠任务」;下一篇解决「真实 HTTP 与大模型 API 怎么发」。五者齐备,脚本才既写得快,又叠得敢——下一篇再给它配上网络请求的实话。
