Python asyncio 深度教程 | Felix’s Micro Space

type

date

slug

1. 并发模型对比

开始前，先厘清 Python 并发生态：

模型	实现方式	适用场景	线程数
并行 (Parallelism)	`multiprocessing`	CPU 密集型（计算、图像处理）	多进程多核
多线程 (Threading)	`threading`	I/O 密集型（文件、网络）	多线程单核
异步 I/O (Async I/O)	`asyncio`	I/O 密集型（高并发网络）	单线程单核

核心区别： asyncio 使用协作式多任务——协程主动让出控制权，而非被操作系统抢占。

2. 异步思维模型

国际象棋大师类比：

Judit Polgár 同时与 24 位业余选手对弈：

- 同步方式： 一盘下完再下一盘 → (55+5) × 30 × 24 = 720 分钟 = 12 小时

- 异步方式： 每桌走一步就换下一桌 → 24 × 5 × 30 = 3600 秒 = 1 小时

Judit 始终是一个人，但异步调度让效率提升 12 倍。

asyncio 就是这个调度器——在协程等待 I/O 时，切换执行其他就绪任务。

3. 事件循环 (Event Loop) 深度解析

事件循环是 asyncio 的核心引擎——它负责调度和执行所有异步任务。

3.1 核心组件关系

组件	职责	类比
Event Loop	单线程调度器，管理所有任务的执行	CPU 调度器
Coroutine	可暂停/恢复的函数，包含实际逻辑	线程的执行体
Task	协程的包装器，追踪状态和结果	进程控制块 (PCB)
Future	占位符，表示尚未完成的结果	Promise

3.2 执行流程

核心机制：

协作式多任务：协程主动让出控制权（通过 await），而非被强制中断

单线程执行：同一时刻只有一个协程在运行，无需锁机制

I/O 多路复用：事件循环监听多个 I/O 事件，就绪即执行

3.3 底层原理：I/O 多路复用

asyncio 的高效源于操作系统的 I/O 多路复用机制：

机制	平台	特点
select	跨平台	最大 1024 fd，每次轮询所有 fd
poll	Unix	无 fd 数量限制，仍需轮询
epoll	Linux	O(1) 事件通知，高并发首选
kqueue	BSD/macOS	类似 epoll，支持更多事件类型

为什么单线程能处理高并发？

传统多线程：每个连接一个线程，10000 连接 = 10000 线程（内存爆炸）

asyncio：单线程 + epoll 监听 10000 fd，只在有数据时才唤醒协程

3.4 事件循环调度机制

3.4.1 epoll 只能处理 fd 事件

epoll 本身只能监听文件描述符 (fd) 的 I/O 就绪状态，无法直接处理定时器、锁、事件等非 I/O 操作。asyncio 不完全依赖 epoll，而是组合多种机制：

3.4.2 事件循环内部结构

await 类型	底层机制	唤醒方式
网络 I/O	epoll 监听 fd	fd 就绪 → 回调加入 `_ready`
`asyncio.sleep()`	定时器堆 `_scheduled`	时间到期 → 回调加入 `_ready`
Lock/Event/Queue	Future + 内部回调链	`set_result()` → 回调加入 `_ready`
`run_in_executor()`	线程池 + self-pipe	线程完成 → 写 pipe 唤醒 epoll

3.4.3 事件循环核心流程

关键洞察：

epoll 的 timeout 参数是关键：通过设置合适的超时，epoll_wait 会在最近定时器到期时自动返回

所有唤醒最终都变成回调：无论是 I/O、定时器还是 Future，完成后都是把回调加入 _ready 队列

协程状态保存在 Task 对象中：await 时协程挂起，回调触发时调用 coro.send() 恢复

3.4.4 跨平台 Selector 自动选择

Python 通过 selectors 模块抽象了不同操作系统的 I/O 多路复用机制，自动选择当前系统最优的实现：

操作系统	DefaultSelector 选择	备选
Linux 2.5.44+	`EpollSelector`	PollSelector → SelectSelector
macOS / BSD	`KqueueSelector`	PollSelector → SelectSelector
Windows	`SelectSelector`	（或 ProactorEventLoop 用 IOCP）
Solaris	`DevpollSelector`	PollSelector → SelectSelector

查看当前使用的 Selector：

Windows 特殊情况：

Windows 上 asyncio 有两种事件循环：

3.5 启动与获取循环

4. 协程 (Coroutine) 深度解析

协程是 asyncio 的基本执行单元——一个可以暂停和恢复执行的函数。

4.1 协程 vs 普通函数

4.2 await 语义详解

await 是协程的挂起点，它告诉事件循环：“我要等待这个操作，你可以去执行其他任务”。

await 触发场景分类

await 不仅仅用于 I/O 操作！ 它可以用于任何实现了 __await__ 的 awaitable 对象：

类别	示例	等待什么
网络 I/O	`await session.get(url)`	socket 数据就绪
文件 I/O	`await` `aiofiles.open()`	文件读写完成
定时器	`await asyncio.sleep(1)`	时间流逝（非 I/O）
其他协程	`await other_coro()`	协程执行完毕
Task	`await task`	任务完成
锁/信号量	`await lock.acquire()`	获取锁成功
事件	`await event.wait()`	事件被 set()
队列	`await queue.get()`	队列有数据
子进程	`await process.wait()`	进程退出
线程池	`await run_in_executor(...)`	线程任务完成
Future	`await future`	结果被设置

核心理解：await X = “我要等 X 完成，先去忙别的吧”。触发条件是 X 实现了 __await__ 方法，不限于 I/O，任何“需要等待”的场景都可以。

4.3 Awaitable 对象类型

类型	创建方式	特点
Coroutine	`async def func()`	基本执行单元，可暂停/恢复
Task	`asyncio.create_task()`	协程包装器，可取消/查询状态
Future	`loop.create_future()`	低级占位符，一般不直接使用

4.4 Task vs Coroutine

4.5 关键规则总结

规则	说明	示例
`async def` 定义协程函数	调用返回协程对象，不立即执行	`coro = async_func()`
`await` 只能在 `async def` 内	在普通函数中使用会 SyntaxError	`await coro`
`await` 只接受 awaitable	协程、Task、Future 或实现 `__await__` 的对象	`await asyncio.sleep(1)`
未被 await 的协程不会执行	会触发 RuntimeWarning	需 `await` 或 `create_task`
阻塞调用会冻结事件循环	`time.sleep()` 等同步 I/O 会阻塞所有任务	用 `asyncio.sleep()` 替代

5. 常用 API 分类

5.1 任务创建与调度

5.2 同步原语

5.3 异步队列

6. 编程模式

6.1 协程链 (Coroutine Chaining)

适用于有依赖关系的顺序任务：

6.2 生产者-消费者模式

适用于解耦数据生产与处理：

7. 高级特性

7.1 异步迭代器与生成器

7.2 异步上下文管理器

7.3 异常处理 (Python 3.11+)

8. 常见问题解决方案

8.1 任务取消机制

最佳实践：

始终在 except CancelledError 中执行清理

清理完成后重新 raise，确认取消状态

使用 asyncio.shield() 保护关键操作不被取消

8.2 超时控制

方式	Python 版本	特点
`wait_for()`	3.7+	简单，但超时后任务被取消
`timeout()`	3.11+	上下文管理器，更灵活
`timeout_at()`	3.11+	绝对时间截止，适合多步骤操作

8.3 混合同步/异步代码

当必须使用同步库时，用 run_in_executor 避免阻塞事件循环：

适用场景：

CPU 密集型任务（图像处理、加密）

无异步版本的第三方库

文件 I/O（如果不用 aiofiles）

8.4 多线程与 asyncio 协作

从其他线程向事件循环提交协程：

常见应用： GUI 应用 + asyncio 后台、Web 服务 + 工作线程

8.5 优雅关闭模式

9. 常见陷阱与最佳实践

陷阱	原因	解决方案
协程未执行	`coro()` 未 `await` 或 `create_task`	确保所有协程被调度
阻塞调用	`time.sleep()`, 同步 I/O	用 `asyncio.sleep()`, `aiohttp`
任务泄漏	`create_task()` 后未保存引用	保存引用或用 `TaskGroup`
回调地狱	过度使用 `add_done_callback`	优先用 `await` 线性写法
并发限制	无限并发导致资源耗尽	使用 `Semaphore` 限流

Python 3.11+ 推荐：TaskGroup

10. 何时使用 asyncio

✅ 适用场景

网络 I/O： HTTP 请求、WebSocket、数据库查询

高并发服务： Web API、聊天服务器、爬虫

Fire-and-forget： 日志上报、消息推送

❌ 不适用场景

CPU 密集型： 图像处理、科学计算 → 用 multiprocessing

阻塞库： 无异步版本的第三方库 → 用 run_in_executor

简单脚本： 无并发需求 → 同步代码更简单

11. 生态库推荐

类别	库	说明
Web 框架	FastAPI, Starlette, Sanic	高性能异步 Web 框架
HTTP 客户端	aiohttp, httpx	异步 HTTP 请求
数据库	asyncpg, motor, databases	异步数据库驱动
文件 I/O	aiofiles	异步文件读写
测试	pytest-asyncio	pytest 异步支持

12. 调试与测试

12.1 调试模式

调试模式会检测：

慢回调（执行超过 100ms 的同步代码）

未被 await 的协程

未关闭的资源

12.2 日志配置

12.3 任务状态检查

12.4 追踪未等待的协程

12.5 使用 pytest-asyncio 测试

pytest.ini 配置：

12.6 性能分析

参考资料

Real Python: Async IO in Python

Python 官方文档: asyncio