Python asyncio 异步编程完全指南：从事件循环到生产实战
写下这篇文章之前，我先坦白一件事：我很长一段时间都在回避 asyncio。心里清楚它是处理高并发 I/O 的大杀器，可在代码里真的去用 async/await，总感觉脑子里那根弦搭不上——为什么这个函数要加 async，为什么 await 不能随便写，为什么别人的协程跑得飞快，我的却动不动就卡死。
直到我在一个爬虫项目里，用同步 requests 抓 200 个页面，等了快十分钟，同事用 asyncio 加 aiohttp 三十秒搞定。那一刻的落差让我下定决心把它啃下来。这篇东西，就是我从坑里爬出来的笔记，写给同样卡在这里的同行。
先搞清楚：协程到底解决了什么问题
很多人一听「异步」就想到多线程，其实这是两码事。多线程靠操作系统切换线程，切换本身有开销，还得防着共享数据的竞争；而 asyncio 的协程是在单线程里靠事件循环调度的，切换的代价小得多，也没有锁的大多数烦恼。
它的适用场景很明确：I/O 密集、等待时间长的任务。网络请求、数据库查询、文件读写、调用外部 API，这些都是等着对方返回结果，CPU 基本在闲着。这种时候把等待的时间拿去做别的事，就是异步的价值。
反过来说，如果你的任务是纯计算的——比如算圆周率、跑矩阵乘法——那 asyncio 帮不上忙，还得多线程或多进程来并行。
理解这一点，比背任何语法都重要。不光是 asyncio，之前写 Go 的 goroutine 也是同一个思路：让出等待的时间。
从零开始：一个能跑的最小例子
先别管那些高级概念。看这段：
import asyncio
async def fetch(url):
print(f"开始请求 {url}")
await asyncio.sleep(1) # 模拟一次网络 I/O
print(f"完成请求 {url}")
return f"{url} 的内容"
async def main():
# 并发跑三个任务，而不是一个个等
results = await asyncio.gather(
fetch("https://a.com"),
fetch("https://b.com"),
fetch("https://c.com"),
)
print(results)
asyncio.run(main())
跑一遍你会发现，三个「请求」几乎是同时完成的，总共只花了约 1 秒，而不是 3 秒。关键就在 await asyncio.sleep(1) 这里——它把控制权让回给了事件循环，让循环有机会去调度别的协程。
几个基础点先记住：
async def定义的函数叫协程函数，调用它不会执行函数体，而是返回一个协程对象。await只能写在async函数内部（Python 3.7+ 也有asyncio.run这个总入口）。asyncio.run(main())是运行协程的标准入口，它会创建一个事件循环，跑完再关掉。
事件循环：一切的调度核心
事件循环（event loop）是 asyncio 的心脏。它维护一个任务队列，不断地取任务执行，遇到 await 就挂起当前任务、去执行别的就绪任务，如此循环。
# 3.10 之前的写法
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
# 3.7+ 已推荐直接这样
asyncio.run(main())
现在绝大多数场景用 asyncio.run() 就够了。但有一个经典陷阱得提：别在一个已经在运行的事件循环里再 asyncio.run()。比如你在 Jupyter Notebook 里反复调用，或者在 FastAPI 的路由函数里又调 run，都会报 RuntimeError: This event loop is already running。Notebook 里要用 await 直接调用，或者用 nest_asyncio 打补丁。
创建任务：create_task 和 gather 的区别
这是新手最容易混的一对。看例子：
async def main():
# 方式一：gather —— 一次性并发跑完所有协程
results = await asyncio.gather(
fetch("a"), fetch("b"), fetch("c")
)
# 方式二：create_task —— 手动把协程包成任务，交给循环
task1 = asyncio.create_task(fetch("a"))
task2 = asyncio.create_task(fetch("b"))
task3 = asyncio.create_task(fetch("c"))
result1 = await task1
result2 = await task2
result3 = await task3
两者的结果一样，但语义不同。gather 适合「一批任务一起跑、一起收结果」；create_task 适合你需要更细粒度控制——比如某个任务跑一半，你想先做点别的，或者想在一批任务里随时取消掉一个。
create_task 有个常被忽略的点：创建任务后一定要保存引用。如果任务对象被垃圾回收了，任务可能永远不被执行，你也不会收到任何报错，排查起来很痛苦。
超时与取消：别让协程悄悄挂死
真实项目里，网络请求动不动就超时。如果我上面的 fetch 换成一个真的卡住的请求，整个程序就会一直等下去。所以超时控制必不可少：
try:
result = await asyncio.wait_for(fetch("https://slow.com"), timeout=5)
except asyncio.TimeoutError:
print("请求超时了")
wait_for 超时会抛出 asyncio.TimeoutError，同时取消那个协程。注意它和 except TimeoutError（内置的）是不一样的，别写混。
要取消一个任务，用 .cancel()：
task = asyncio.create_task(fetch("https://slow.com"))
await asyncio.sleep(0.5)
task.cancel() # 主动取消
try:
await task
except asyncio.CancelledError:
print("任务已取消")
取消之后 await 那个任务会抛 CancelledError。这个异常在 Python 3.8 起继承自 BaseException 而非 Exception，所以 except Exception 是接不住它的——这也是个容易踩的点。
实战：用 aiohttp 写个并发爬虫
空谈概念没意思，直接上一个能用的例子。我们要并发抓多个 URL，还要限制并发数别把对方服务器打爆：
import asyncio
import aiohttp
from asyncio import Semaphore
async def fetch_page(session, url, sem):
async with sem: # 控制并发数
try:
async with session.get(url, timeout=10) as resp:
body = await resp.text()
print(f"{url} -> {len(body)} 字节")
return body
except asyncio.TimeoutError:
print(f"{url} 超时")
return None
async def main():
urls = [f"https://httpbin.org/get?id={i}" for i in range(30)]
sem = Semaphore(5) # 最多同时 5 个请求
async with aiohttp.ClientSession() as session:
tasks = [
asyncio.create_task(fetch_page(session, url, sem))
for url in urls
]
results = await asyncio.gather(*tasks, return_exceptions=True)
good = [r for r in results if r]
print(f"成功 {len(good)} 个")
asyncio.run(main())
几个要点：
Semaphore(5)是信号量，async with sem保证同一时间最多 5 个请求在跑。没有它，你 30 个请求会瞬间全打出去，容易触发对方限流甚至封 IP。return_exceptions=True让gather即使某个任务抛异常，也不会影响到其它任务的结果。async with aiohttp.ClientSession()是整个 session 的生命周期都复用的连接池，别每个请求都新建 session，那样连接复用就白费了。
生产中常见的几个坑
啃完基础，下面是我自己踩过、并且看到别人反复踩的坑，值得单独拎出来：
1. 在同步代码里调用协程函数，得到一堆 warning
async def fetch(url):
...
# 错误：这样调用根本不会执行
result = fetch("https://a.com")
# RuntimeWarning: coroutine 'fetch' was never awaited
必须用 await 或者 asyncio.run/create_task 去驱动它。这是新手最常遇到、却最容易被忽略的 warning。
2. 阻塞调用混进协程
这是 asyncio 里的「隐形杀手」。如果你在协程里调用了 time.sleep()、requests.get()、open() 大文件这类阻塞的同步操作，整个事件循环都会被卡住，其它所有协程都得等着：
import time
async def bad():
time.sleep(5) # 这是阻塞的！会把整个循环卡住
async def good():
await asyncio.sleep(5) # 这才是让出控制权
记住一条铁律：协程里凡是会「等」的地方，都要用支持异步的版本——asyncio.sleep、aiohttp、asyncpg、aiofiles，而不是它们的同步兄弟。如果实在要用同步库，用 await asyncio.to_thread(...) 把它丢到线程池去跑。
3. 异常被吞掉
results = await asyncio.gather(*tasks)
如果某个任务抛异常，gather 默认会立刻把异常抛出来打断你。想要「一个坏了不影响全部」，记得加 return_exceptions=True。
什么时候该用，什么时候别硬上
最后给个实用的判断清单：
- 数据抓取、API 聚合、消息队列消费：异步的完美主场。
- Web 框架（FastAPI、Sanic、aiohttp 服务端）：天生异步，放心用。
- 纯 CPU 计算：别用 asyncio，上多进程或
concurrent.futures.ProcessPoolExecutor。 - 已经在用多线程且工作良好：不必为换而换。异步代码的调试和排查本来就比顺序代码难，收益不明显的场景，保持简单更好。
asyncio 不难，难的是观念的转变——从「让别人替我并行做事」变成「在等待的时候我不闲着」。这个念头转过来了，那些 async、await 的语法，其实就只是照章办事而已。
希望这篇对你有用。如果哪个地方你没看明白，或者我哪里写错了，欢迎在评论区拍砖。