asyncio 最容易踩的 5 个坑，第 4 个我研究了一下午

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

asyncio 最容易踩的 5 个坑，第 4 个我研究了一下午

把同步代码改成 asyncio，听着挺简单，run、await、create_task 这几个词往代码里一塞就完事了。但真改起来，你会发现事情没想的那么顺利。

我整理了一下我踩过的坑，都是网上资料不多提、但又特别容易栽跟头的。

坑 1：在同步函数里调 await

新手最常犯的错。代码长这样：

def process_data(url):
resp = requests.get(url) # 同步阻塞
return resp.json()

然后你说，没事，我换成 asyncio 试试：

async def process_data(url):
resp = await session.get(url) # 这次是协程
return resp.json()

await 关键字必须在 async def 函数里用。如果你写的是普通 def，里头 await 直接报语法错误。

但反过来呢？你在 async 函数里调了同步 IO，照样阻塞整个事件循环。比如：

async def handler(request):
data = open("big.txt").read() # 同步 IO，事件循环被卡住
return web.Response(text=data)

这一行 read 会把整个事件循环停住，期间所有别的协程都得等它读完。正确做法是用 aiofiles，或者把同步操作丢到线程池里：

loop = asyncio.get_event_loop()
data = await loop.run_in_executor(None, open("big.txt").read)

坑 2：忘了关 ClientSession

aiohttp 的 ClientSession 用完不关，连接池里的连接会一直挂着，直到事件循环结束。

async def fetch(url):
async with aiohttp.ClientSession() as session: # ✅ 用 async with
async with session.get(url) as resp:
return await resp.text()

如果你写的是：

session = aiohttp.ClientSession()
# … 用完没关

那这个 session 里的连接会留到进程退出才释放。如果你的脚本是长跑的，连接越积越多，最后把系统文件描述符耗光。

坑 3：gather 里一个报错，全部一起挂

results = await asyncio.gather(
fetch(url1),
fetch(url2),
fetch(url3),
)

如果 url2 那个 fetch 抛了异常，默认情况下 gather 会立刻把异常抛出来，url1 和 url3 的结果直接被丢弃——哪怕它们已经成功了。

要避免这个问题，加 return_exceptions=True：

results = await asyncio.gather(
fetch(url1), fetch(url2), fetch(url3),
return_exceptions=True
)
# results 里要么是结果，要么是异常对象
for r in results:
if isinstance(r, Exception):
log(r)
else:
use(r)

坑 4：事件循环只能跑一次

这事我栽过一次，搞了一下午才明白。

import asyncio

async def hello():
print("hello")
await asyncio.sleep(1)
print("world")

asyncio.run(hello())
asyncio.run(hello()) # 这一行会报 "RuntimeError: Event loop is closed"

asyncio.run() 会创建一个新的事件循环，跑完你给的协程，再关掉这个循环。第二次调用 asyncio.run 时，它试图在新线程里复用已经关闭的循环，直接报错。

如果你想"同一份代码跑两遍"——比如某些测试场景——别在模块顶层写 asyncio.run()，应该把它包在 if __name__ == "__main__": 里，或者写一个主函数手动管循环。

更隐蔽的版本是 Jupyter Notebook。Notebook 里有自己的一套事件循环，你直接 asyncio.run() 会和它冲突，要么用 nest_asyncio，要么用 await 直接跑（Notebook 顶层 await 是支持的）。

坑 5：CPU 密集任务丢给 asyncio

asyncio 适合 IO 密集场景。如果你的任务是 CPU 密集的（比如图像处理、压缩、加密），丢给 asyncio 反而更慢——因为一个协程占着 CPU 不让出，其他协程全得等。

这种任务正确做法是用 ProcessPoolExecutor：

from concurrent.futures import ProcessPoolExecutor

async def heavy_computation(data):
loop = asyncio.get_event_loop()
with ProcessPoolExecutor() as pool:
result = await loop.run_in_executor(pool, cpu_heavy_func, data)
return result

run_in_executor 把任务丢到进程池里执行，主进程的事件循环不会被卡住。这套写法 IO 密集和 CPU 密集都能 cover。

最后

asyncio 是个好工具，但它的"简单"是有条件的。你需要：

• async 函数里不能有同步阻塞 IO
• 资源（session、连接、文件）要记得关
• 异常处理要明确，不然一个错全盘都错
• 选对场景——IO 密集用它，CPU 密集用进程

把这些想清楚，asyncio 才能真的帮你提效，不然就是给自己挖坑。