异步编程 asyncio
校招考察频度：★ 高频（核心必背·高频生死线） 一句话速记：asyncio 是 Python 突破单线程 I/O 瓶颈的物理外挂，也是现代 FastAPI 与 AI 智能体编排的物理地基；其本质是用户态协作式调度 + Linux epoll 多路复用；协程通过
await主动让出 CPU 控制权，严禁任何同步阻塞调用（必须用to_thread丢入线程池隔离）。
在 Python 后端与大模型面试中，面试官对并发的考察从早期的“死背 GIL”全面转向了“asyncio 底层事件循环是如何工作的？为什么你的 FastAPI 吞吐量能翻数倍？为什么有人用了 async 接口反而变得更卡了？”
一、0 基础生动类比：单人厨房做大餐模型
如果把高并发网络请求比作**“一位厨师为 100 桌客人做大餐”**：
| 并发模型 | 现实做饭场景 | 优缺点分析 |
|---|---|---|
| 传统多线程模型 | 招聘 100 位厨师各守一个灶台，各做各的菜。 | 只要有炉灶等待（如慢火炖汤），厨师就在那干瞪眼。100 位厨师占满厨房（内存暴涨、操作系统线程上下文切换打死 CPU）。 |
| asyncio 协程事件循环 | 只有 1 位身手敏捷的米其林大厨（单线程），配一台带定时提醒的蜂鸣器（epoll 事件循环）。 | 大厨把汤炖在 1 号锅上，按好定时器（await 让出控制权），一秒钟都不闲着，立刻转身去 2 号锅炒菜；当 1 号锅炖好蜂鸣器响起，大厨再转回来出锅。1 个人玩转 100 桌！ |
二、底层运行机制：从生成器 yield 到原生协程
asyncio 协作式单线程调度时序
1. 协程的演进本质
- Python 2.x/3.4 之前：通过生成器的
yield和yield from实现暂停与恢复，代码晦涩难懂。 - Python 3.5+：引入原生的
async def（定义协程）与await（挂起让出）。 - 执行规则：调用
async def函数不会立即执行任何代码，它仅仅返回一个“协程对象（coroutine object）”，必须放入事件循环中（通过await或asyncio.create_task）才会被真正排期调度。
2. await 的严格约束
await只能出现在async def函数内部。await后面的对象必须是 Awaitable（可等待对象），分为三种：- Coroutine（协程对象）
- Task（任务）：对协程的包装，放入循环立即启动。
- Future（未来对象）：底层低级对象，代表一个尚未完成的异步操作结果。
三、高并发控制工具箱：Task、Gather 与信号量
import asyncio
import httpx
# 1. 限制最大并发量为 10 (防止打爆下游目标服务器或被封禁)
sem = asyncio.Semaphore(10)
async def fetch_url(client: httpx.AsyncClient, url: str) -> str:
async with sem: # 争抢令牌，超出 10 个自动挂起排队
resp = await client.get(url, timeout=5.0)
return resp.text
async def main():
urls = [f"https://api.example.com/items/{i}" for i in range(100)]
async with httpx.AsyncClient() as client:
# 并发创建 100 个 Task 并通过 gather 聚合等待
tasks = [fetch_url(client, u) for u in urls]
# 并发执行，收集所有结果
results = await asyncio.gather(*tasks, return_exceptions=True)
print(f"成功抓取 {len(results)} 条数据")
# 启动事件循环
asyncio.run(main())| 工具 | 核心职责 | 避坑提醒 |
|---|---|---|
asyncio.create_task() | 将协程包装为 Task 并立即提交给事件循环调度（非阻塞，后台并发跑）。 | 必须保留其对象引用，否则可能被垃圾回收器意外回收。 |
asyncio.gather(*tasks) | 并发等待一批任务，按传入顺序返回结果列表。 | 建议加上 return_exceptions=True，防止单个任务报错导致全部崩溃。 |
asyncio.Semaphore(n) | 信号量并发限制器。限制同一时刻执行临界区的协程数量。 | 爬虫与微服务外呼必配，防止耗尽系统文件句柄或下游熔断。 |
四、致命隐患：同步阻塞卡死事件循环与排障
初学者常犯的弥天大祸：在 async def 协程函数里写了阻塞式同步代码！
# ❌ 错误示范：一粒老鼠屎坏了一锅粥！
async def bad_handler():
import time
time.sleep(5) # 致命！直接卡死主线程 5 秒！
# 导致全服务所有其他成千上万个协程在这个 5 秒内全盘冻结，无法响应任何新请求！为什么会卡死？
因为 asyncio 的事件循环是单线程的。协程的切换全靠代码在 I/O 时主动通过 await 让出控制权（协作式多任务）。如果调用了 time.sleep()、同步 requests.get()、同步数据库驱动或密集的 CPU 大计算，主线程被操作系统霸占，事件循环根本无法推进，导致全站“假死”！
治本良方：三大排障方案
- 换用异步库（最佳）：
- 替换
requests➔httpx.AsyncClient - 替换
time.sleep➔await asyncio.sleep - 替换
pymysql➔asyncmy/aiomysql
- 替换
- 使用
asyncio.to_thread隔离同步调用（Python 3.9+ 推荐）：# 把遗留的同步阻塞函数，优雅抛到后台底层的独立线程池中执行 result = await asyncio.to_thread(legacy_blocking_io_function, arg1, arg2) - 密集 CPU 计算抛入进程池：
- 使用
loop.run_in_executor(ProcessPoolExecutor, cpu_heavy_task)，彻底绕过 GIL 利用多核。
- 使用
五、知识脉络关联网（Knowledge Highway）
- ⬅️ 底层系统支撑：asyncio 的事件循环在 Linux 下底层基于 epoll（I/O 多路复用）非阻塞事件通知机制。
- ➡️ 现代 Web 落地：FastAPI 是如何利用 asyncio 实现数万并发长连接的？参见 FastAPI 异步高并发与 ASGI 架构。
- 🔄 前沿 AI 智能体应用：大模型流式输出（SSE）与多工具异步并发调用实战，参见 AI 智能体流式输出与异步实践。
六、校招高频面试追问
- 既然 Python 有 GIL，为什么 asyncio 还能实现高并发？
- GIL 限制的是多核真并行计算，而限制不了单核的 I/O 多路复用。
- 在高并发网络服务中（如 Web API），99% 的时间都在等待数据库查询、Redis 响应或第三方 RPC 传输（I/O 密集型）。asyncio 在单线程内借助
epoll管理海量 Socket，当一个请求在等待网络 I/O 时，CPU 迅速去处理下一个请求。由于全在用户态切换，没有线程上下文切换和锁竞争开销，因此吞吐量呈几何级暴涨。
asyncio.run()的底层做了什么？可以在已经运行的循环里再次调用吗？asyncio.run()会新建一个全新的事件循环、将传入的协程封装为 main 任务执行、运行所有未完成的任务，并在结束后优雅关闭循环与线程池。- 严禁在已有运行事件循环的环境中再次调用
asyncio.run()（会抛出RuntimeError: This event loop is already running，常见于 Jupyter Notebook 或 Tornado/FastAPI 内部），此时应使用asyncio.create_task()或直接await。