企业官网建设流程全解析

为什么你的 async/await 没让代码变快？Python 异步编程的 5 个真实陷阱

前言：写上 async，不等于获得了异步

如果你写过 Python 异步代码，大概率遇到过这样一种让人有点怀疑人生的结果：

代码从同步改成了 async/await，运行时间却没有明显下降。更尴尬的是，有时候它还变慢了。

网络请求场景确实快了一些，但远没有“并发之后应该飞起来”的感觉。于是你开始怀疑：是不是 asyncio 不行？是不是 await 用错了？还是 Python 的异步本来就只是语法上的装饰？

通常都不是。

更常见的原因是：代码看起来是异步的，执行方式却仍然是同步的。

asyncio 并不是一个“加速按钮”。它擅长的是在任务等待 I/O 时，把执行机会让给其他任务。网络请求、数据库查询、文件读写等操作在等待外部资源时，CPU 往往并不忙；异步的价值，就是利用这段等待时间去推进别的工作。

如果代码里没有真正的异步等待，或者等待的方式不对，async def 只会给同步逻辑换一件新外套。下面这 5 个陷阱都很常见：代码能运行，结构也像异步，性能却没有跟上。

陷阱一：在 async 函数里调用同步阻塞代码

这是最经典，也最容易让整个事件循环停摆的错误。

假设我们要同时请求 10 个地址：

importasyncioimportrequestsasyncdeffetch(url):response=requests.get(url)# 同步阻塞调用returnresponse.json()asyncdefmain():urls=["https://example.com"]*10awaitasyncio.gather(*(fetch(url)forurlinurls))asyncio.run(main())

这段代码有 async def，也有 asyncio.gather()，看上去相当有异步气质。可真正执行请求的 requests.get() 是同步阻塞调用。

当第一个 requests.get() 等待服务器响应时，事件循环也被占住了。其他协程没有机会运行，只能排队等前一个请求结束。于是 gather() 表面上把任务放在了一起，实际效果却接近“一个接一个地执行”。

修复方式：使用真正的异步 HTTP 客户端

importasyncioimportaiohttpasyncdeffetch(session,url):asyncwithsession.get(url)asresponse:returnawaitresponse.json()asyncdefmain():urls=["https://example.com"]*10asyncwithaiohttp.ClientSession()assession:tasks=[fetch(session,url)forurlinurls]awaitasyncio.gather(*tasks)asyncio.run(main())

这里的关键不是把 requests 换成了一个名字里带有 async 的库，而是 session.get() 本身能够在等待网络响应时挂起当前协程，让事件循环继续处理其他任务。

需要特别留意的同步阻塞操作包括：

• requests.get() 等同步 HTTP 请求；
• time.sleep()；
• 同步数据库查询；
• 同步文件读写；
• 会长时间占用 CPU 的普通函数调用。

一个简单的检查方法是：看 async def 内部的等待是否真的来自异步 API。 如果函数里只是写了 async def，却没有等待任何真正的异步操作，它很可能只是“长得像异步的同步函数”。

陷阱二：在循环里逐个 await，把并发写成了串行

第二个陷阱正好相反：不是没有 await，而是每一步都立刻 await 了。

看这段代码：

asyncdefprocess_users(user_ids):results=[]foruser_idinuser_ids:user=awaitfetch_user(user_id)results.append(user)returnresults

这段代码当然是合法的异步代码，但它会等待第一个用户请求完成后，才开始第二个请求。

假设有 100 个用户，每个请求平均需要 100 毫秒，且服务端和连接池都允许并发，那么这种写法的总耗时可能接近 10 秒。你虽然使用了异步函数，却把请求排成了一列，异步的优势也就被自己收回去了。

修复方式：先准备任务，再统一等待

importasyncioasyncdefprocess_users(user_ids):tasks=[fetch_user(user_id)foruser_idinuser_ids]returnawaitasyncio.gather(*tasks)

这样，多个请求可以同时进入等待状态，总耗时通常更接近最慢的那个请求，而不是所有请求耗时之和。

不过这里有一个容易被忽略的细节：并发不等于无限制地同时发起任务。如果 user_ids 有几十万条，直接创建几十万个任务也可能造成内存压力、连接数过多，甚至触发对方服务的限流。

实际项目中，通常需要配合并发上限，例如使用 asyncio.Semaphore：

importasyncioasyncdefprocess_one(user_id,semaphore):asyncwithsemaphore:returnawaitfetch_user(user_id)asyncdefprocess_users(user_ids,limit=20):semaphore=asyncio.Semaphore(limit)tasks=[process_one(user_id,semaphore)foruser_idinuser_ids]returnawaitasyncio.gather(*tasks)

这里的原则可以概括成一句话：

await 是等待，不是启动。想要并发，就先创建多个任务，再统一等待它们。

如果你希望“谁先完成就先处理谁”，可以考虑 asyncio.as_completed()；如果希望收集所有结果，asyncio.gather() 通常更直观。

陷阱三：没有设置超时，让一个请求拖住整批任务

异步程序通常会处理更多并发请求，因此超时不是锦上添花，而是基本的可靠性设置。

下面这段代码的问题不在语法，而在于它没有明确的等待上限：

asyncdeffetch(session,url):asyncwithsession.get(url)asresponse:returnawaitresponse.json()

如果目标服务器响应很慢、网络中间设备出了问题，或者连接建立后对方一直不返回数据，这个协程就可能等待很久。

需要澄清的是：一个协程卡住，并不一定意味着整个事件循环都停止了。其他协程仍然可能继续运行。但在批量任务中，只要你在等待一个整体性的 gather()，这个迟迟不结束的任务就可能拖住最终结果，让调用方一直等下去。

修复方式：为单次请求设置超时

以 aiohttp 为例，可以为客户端配置超时：

importasyncioimportaiohttpasyncdeffetch(session,url):try:asyncwithsession.get(url)asresponse:returnawaitresponse.json()exceptasyncio.TimeoutError:returnNoneasyncdefmain():timeout=aiohttp.ClientTimeout(total=10)asyncwithaiohttp.ClientSession(timeout=timeout)assession:returnawaitfetch(session,"https://example.com")

也可以在更高一层为整批任务设置总超时：

results=awaitasyncio.wait_for(asyncio.gather(*tasks),timeout=30,)

这两个超时解决的问题不完全相同：

• 单请求超时：限制某一个请求最多等待多久；
• 整批任务超时：限制这一批工作最多占用多久。

生产环境里通常还需要考虑取消任务、重试策略、错误分类和日志记录。超时后直接返回 None 很简单，但不一定适合所有业务。对于关键数据，最好明确区分“请求失败”“请求超时”和“服务端返回了空结果”，否则后续排查会变成一场小型考古活动。

陷阱四：把 CPU 密集任务交给 asyncio

这是一个方向上的误用。

asyncio 适合处理 I/O 等待，而不是让 CPU 计算自动变快。

例如：

importasyncioasyncdefcompute(data):# 复杂的计算逻辑returnsum(x**2forxinrange(10_000_000))asyncdefmain(dataset):returnawaitasyncio.gather(*(compute(data)fordataindataset))

这段代码虽然使用了 async def 和 gather()，但 compute() 内部没有任何 I/O 等待。它一旦开始计算，就会一直占用事件循环所在的线程。

事件循环不是魔法管家，不能在一段纯 Python 计算进行时凭空把 CPU 让出来。结果就是：一个协程在计算，其他协程只能等着；gather() 并不会把 CPU 计算自动变成并行计算。

修复方式：把 CPU 密集工作移出事件循环

对于计算量较大的任务，可以使用进程池：

importasynciofromconcurrent.futuresimportProcessPoolExecutordefcompute_sync(data):returnsum(x**2forxinrange(10_000_000))asyncdefcompute_async(data,executor):loop=asyncio.get_running_loop()returnawaitloop.run_in_executor(executor,compute_sync,data,)asyncdefmain(dataset):withProcessPoolExecutor()asexecutor:tasks=[compute_async(data,executor)fordataindataset]returnawaitasyncio.gather(*tasks)

实际选择可以先按下面的方向判断：

• I/O 密集：优先考虑 asyncio；
• CPU 密集：优先考虑多进程、ProcessPoolExecutor 或专门的计算框架；
• 既有 I/O，又有计算：让 asyncio 负责 I/O，把重计算交给进程池；
• 轻量级、会阻塞的同步操作：可以考虑线程池，但要注意线程安全和线程数量。

这里还要补充一点：不是所有 CPU 操作都值得立刻上多进程。进程之间传递数据需要序列化，启动和调度也有成本。如果任务很小，进程池的开销可能比计算本身还大。正确做法不是看到 for 循环就上进程池，而是先测量，再决定。

判断标准很简单：你的协程是在等待外部资源，还是一直在做计算？ 前者可能适合异步，后者通常需要换一种并行方式。

陷阱五：业务用了异步，底层库却还是同步的

这是最容易出现在真实项目里的陷阱。

项目的路由函数是异步的，业务逻辑也用了 await，但某个数据库驱动、缓存客户端或第三方 SDK 仍然是同步版本。于是，阻塞只是藏得更深了。

例如：

importpsycopg2asyncdefget_user(user_id):conn=psycopg2.connect(...)cursor=conn.cursor()cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,),)returncursor.fetchone()

这个函数从定义上看是异步的，但 psycopg2.connect()、execute() 和 fetchone() 都是同步操作。每次调用它，都会在当前线程里阻塞事件循环。

修复方式：使用异步驱动

importasyncpgasyncdefget_user(user_id):conn=awaitasyncpg.connect(...)try:returnawaitconn.fetchrow("SELECT * FROM users WHERE id = $1",user_id,)finally:awaitconn.close()

在实际服务中，通常还应该使用连接池，而不是每次查询都创建和关闭一个连接。示例重点是说明：数据库操作本身也必须提供异步等待，不能只把外层函数改成 async def。

常见库可以大致这样对照：

如果确实没有合适的异步版本，可以把同步调用放进线程池，避免直接阻塞事件循环。但这不是把同步代码“变成了异步代码”，而是把它搬到另一个线程执行。线程数、连接数、异常处理和取消行为都需要单独设计。

所以，检查异步项目时不要只看业务函数有没有 await，还要沿着调用链继续往下看：所有 I/O 库是否真的支持异步？

如何自查：先回答这 4 个问题

如果你的异步代码没有明显变快，可以先不要急着重写。按下面的问题逐个排查，通常比凭感觉改代码有效。

1. async 函数里是否调用了同步阻塞操作？

重点检查：

• requests；
• time.sleep()；
• 同步数据库驱动；
• 同步文件操作；
• 没有经过隔离的第三方 SDK。

2. 是否在循环里逐个 await？

如果每次循环都要等前一个任务结束，整体就很可能被写成了串行。确认业务允许并发后，再考虑 gather()、as_completed() 或带上限的任务队列。

3. 每个外部 I/O 是否都有超时？

没有超时的请求，在测试环境里可能一直没暴露问题；到了生产环境，它会用一种非常安静的方式把整个接口拖到超时。

4. 任务本质上是 I/O 密集还是 CPU 密集？

如果主要时间花在等待网络、数据库或磁盘，异步可能很合适；如果主要时间花在压缩、加密、解析或大量计算，asyncio 通常不是解决方案。

这几个问题回答完，瓶颈一般已经露出轮廓了。接下来再用实际耗时、并发数和资源占用做验证，而不是只看代码里出现了多少个 await。

一个更重要的原则：asyncio 不是“让代码变快”的魔法

Python 异步编程的核心，可以浓缩成一句话：

asyncio 是处理 I/O 并发的工具，不是通用加速器。

它的作用是：当一个任务在等待网络、数据库或其他外部资源时，让事件循环有机会推进别的任务。这样，多个 I/O 操作可以重叠等待，整体耗时就不必简单相加。

但它做不到三件事：

• 把同步阻塞调用自动变成异步调用；
• 把纯 CPU 计算自动变成并行计算；
• 在没有控制并发和超时的情况下，替你处理资源耗尽问题。

所以，判断一段代码是否适合异步，看的不是“我希望它变快”，而是：它有多少时间在等待外部资源？

先把这个问题想清楚，再决定要不要写 async。有时候，保持同步代码反而更简单、更稳定；而当任务确实以 I/O 等待为主时，正确使用异步，效果会非常明显。

参考资料

• Python asyncio 官方文档
• aiohttp Client Quickstart
• Python concurrent.futures 官方文档
• asyncpg 官方文档

关于作者

Andy Chu（洪顺） — ParheliaWeb 创始人，荷兰籍华裔开发者。平时写 Python 和 FastAPI，关注出海工具链和 API 工程实践。异步编程里的坑，也踩过不少，所以这篇文章既是经验分享，也是一次集中整理。

GitHub: AndyParhelia