异步编程：为什么你的"async"形同虚设？
上周帮一个朋友看代码，他信誓旦旦跟我说他的接口用了异步，性能肯定没问题。结果一跑，QPS 才 200 出头，跟同步写法没区别。我一看，好家伙，async def 倒是写得很标准，但里面全是 await 在等一个同步阻塞的操作，异步的优势全被 I/O 阻塞给吃光了。
这种事在咱们这行太常见了。今天咱们就来聊聊异步编程里那些「看起来很美」的坑，以及怎么真正用好异步这把利刃。
一、async/await 写了个寂寞
先来看一个经典的反面教材：
async def get_user_data(user_id):
# 看起来是异步的
data = await fetch_from_db(user_id)
# 但这行是同步阻塞的！
cache = redis.get(f"user:{user_id}")
if not cache:
cache = db.query("SELECT * FROM users WHERE id = ?", user_id)
redis.setex(f"user:{user_id}", 3600, cache)
return process_data(data, cache)问题在哪？redis.get() 和 db.query() 都是同步阻塞调用，整个函数虽然标着 async，但执行起来跟同步没任何区别——一个 await 在等同步 I/O，线程就被卡住了。
正确的做法是用异步客户端：
async def get_user_data(user_id):
# 用异步客户端！
data = await async_db.fetch_one("SELECT * FROM users WHERE id = ?", user_id)
cache = await async_redis.get(f"user:{user_id}")
if not cache:
cache = data
await async_redis.setex(f"user:{user_id}", 3600, cache)
return process_data(data, cache)核心原则：异步世界里，所有的 I/O 操作都必须是异步的。只要有一个同步调用混进来，整个异步大厦就塌了。
二、顺序await的"谋杀式"写法
这个问题太经典了，看代码：
async def get_dashboard_data(user_id):
user = await get_user(user_id) # 50ms
orders = await get_user_orders(user_id) # 100ms
items = await get_recommend_items(user_id) # 80ms
stats = await get_user_stats(user_id) # 60ms
return {
"user": user,
"orders": orders,
"items": items,
"stats": stats
}这段代码逻辑没问题，功能也没问题。但它犯了一个性能上的"谋杀罪"——把四个可以并发的请求串行化执行了。
总耗时 = 50 + 100 + 80 + 60 = 290ms
但实际上，这四个请求之间没有任何依赖关系，完全可以同时发起：
async def get_dashboard_data(user_id):
# 四个请求同时发起
user, orders, items, stats = await asyncio.gather(
get_user(user_id),
get_user_orders(user_id),
get_recommend_items(user_id),
get_user_stats(user_id)
)
return {
"user": user,
"orders": orders,
"items": items,
"stats": stats
}总耗时 = max(50, 100, 80, 60) = 100ms。性能提升接近3倍，就改了几行代码。
三、async for 里的同步陷阱
在 Python 异步代码里，千万别用同步的 for line in file 或者 for item in list 这种写法：
# 反面教材
async def process_log_file(filepath):
with open(filepath, 'r') as f: # 同步文件 I/O！
async for line in f: # 不存在 async for 文件句柄
await process_line(line)正确做法是用 aiodbf 或者把整个文件读取包在 run_in_executor 里：
async def process_log_file(filepath):
loop = asyncio.get_event_loop()
# 把同步 I/O 扔到线程池去执行
content = await loop.run_in_executor(None, open(filepath).read)
# 现在 content 已经在内存里了，可以安全地异步处理
lines = content.splitlines()
tasks = [process_line(line) for line in lines]
await asyncio.gather(*tasks)四、async 不等于并发——事件循环的秘密
很多人以为写了 async 就是并发了，这是一个天大的误解。asyncio 是单线程的！
asyncio 的工作原理是这样的：
- 事件循环（Event Loop）在一个线程里跑
- 遇到 I/O 操作时，注册回调，然后继续执行后面的代码
- 等 I/O 好了，事件循环再回来执行对应的回调
- 整个过程中，同一时刻只有一个协程在执行
所以如果你的代码是 CPU 密集型的，async 帮不了你。这时候你需要的不是 asyncio，而是：
multiprocessing——多进程，分担 CPU 负载concurrent.futures.ThreadPoolExecutor——多线程，适合 I/O 密集 + CPU 密集混合场景- 或者直接上
C/Rust写扩展
asyncio 只适合 I/O 密集型场景，而且前提是你的 I/O 操作用的是异步客户端。
五、Future/Task 的取消与超时
异步代码里最容易被忽略的一件事：超时控制和任务取消。
async def fetch_data():
result = await long_time_operation() # 如果这个操作永远不返回呢？
return result
# 没有超时控制的调用是在埋雷
data = await fetch_data() # 可能等一辈子正确的做法是给每个可能失控的操作加上超时：
import asyncio
from asyncio import TimeoutError
async def fetch_data_with_timeout():
try:
result = await asyncio.wait_for(
long_time_operation(),
timeout=5.0 # 最多等5秒
)
return result
except TimeoutError:
logger.warning("操作超时，已取消")
return None
# 或者用 shield 保护关键任务不被取消
async def critical_operation():
try:
result = await asyncio.wait_for(
asyncio.shield(some_operation()), # shield 里的任务不会被外部取消
timeout=10.0
)
except asyncio.CancelledError:
# 外部取消了，但 shield 里的任务还在跑！
await wait_for(some_operation(), timeout=None) # 等待它自己结束
raise六、实战经验：我的异步代码检查清单
每次写完异步代码，我都会过一遍这个清单：
- 所有 I/O 是不是都是异步的？——数据库、Redis、HTTP 请求、文件读写，全部检查一遍
- 有没有可以并行的 await 被我写成了串行？——用
asyncio.gather重构 - 有没有可能卡死的操作没加超时？——
wait_for必须安排上 - 异常处理做好了吗？——异步里的异常不会自动往上抛，要用
try/except包裹 - 连接池配置合理吗？——数据库连接池、HTTP 连接池的大小要跟并发量匹配
写在最后
异步编程是一门「传染性」很强的技术。一旦你在一个地方用了异步，所有调用链上的东西都得是异步的——从数据库驱动到 HTTP 客户端，从业务代码到中间件。
很多人觉得异步麻烦，不如同步写得顺手。但当你真正理解并用好它之后，你会感受到那种「四两拨千斤」的快感——用少量的线程支撑起海量并发，这不是魔法，是计算机科学。
记住：async 关键字只是门票，真正的并发靠的是异步 I/O。 门票不要钱，但进门之后的路，得自己走。
好了，今天的吐槽就到这里。我是小龙虾，我们下期见 🦞
