一、问题背景：一个“看起来很快”的慢接口
上个月接手了一个订单服务的性能优化任务。这个接口的逻辑很简单：根据订单ID查询订单主表，然后依次调用用户服务、商品服务、物流服务获取附加信息，最后拼接返回。
压测结果让人崩溃：QPS刚到200，平均RT就飙到1.8s，P99直接2.8s。而看监控，MySQL的CPU只有25%，Redis的命中率98%，三个下游RPC服务的负载也都很低。典型的“服务端没忙死，请求却慢死”的场景。
用py-spy抓了一下调用栈，发现线程大部分时间阻塞在socket.read上——那就是在等下游HTTP响应。这个接口要串行调用3个下游服务，每个平均耗时400ms，加起来就是1.2s。简直是在用“同步思维”写“异步IO密集”的代码。
二、环境与版本：明确技术栈
先交代一下重构前的技术栈：
- Python 3.9.18（别问我为什么没上3.12，公司基座就这版本）
- Flask 2.2.5 + gunicorn 20.1.0（worker_class=sync, workers=8）
- requests 2.31.0（同步HTTP客户端）
- 下游服务：内部RPC网关，HTTP/1.1，平均RT 350-450ms
重构后引入：
- asyncio 3.9.18（内置）
- httpx 0.25.2（支持异步HTTP）
- uvloop 0.19.0（事件循环加速，实测吞吐再提15%）
- aiohttp 3.9.1（用于连接池监控，和httpx对比测试）
三、方案设计：不换框架，只改IO层
有人说“你直接换FastAPI不就完了”。但业务代码几千行，全部改造风险太大。我的方案是保留Flask同步接口层，内部用asyncio.run()驱动协程并发。
设计要点：
- 用asyncio.gather并发调用3个下游API，替代原来的串行requests
- 用asyncio.Semaphore限制并发数，防止下游服务被冲垮（设为20）
- 用httpx.AsyncClient替代requests，复用HTTP连接池
- 每个worker进程内创建一个事件循环，通过
asyncio.run()在请求处理函数中执行
架构图（文字版）：
Flask(View) --> asyncio.run(main_coroutine)
├── Semaphore(20)
├── gather(
│ fetch_user_info,
│ fetch_product_info,
│ fetch_logistics_info
│ )
└── 合并结果返回
四、核心实现：Before vs After代码
Before：同步串行代码
# app.py - 重构前
import requests
from flask import Flask, jsonify
app = Flask(__name__)
def fetch_user_info(order):
# 模拟下游RPC调用，耗时400ms
resp = requests.get(f"http://user-service/api/user/{order['uid']}", timeout=1.5)
return resp.json()
def fetch_product_info(order):
resp = requests.get(f"http://product-service/api/product/{order['pid']}", timeout=1.5)
return resp.json()
def fetch_logistics_info(order):
resp = requests.get(f"http://logistics-service/api/logistics/{order['oid']}", timeout=1.5)
return resp.json()
@app.route("/api/order/")
def get_order(order_id):
# 模拟从MySQL查询订单主表（耗时50ms）
order = {"oid": order_id, "uid": 123, "pid": 456}
user = fetch_user_info(order) # 400ms
product = fetch_product_info(order) # 400ms
logistics = fetch_logistics_info(order) # 400ms
return jsonify({
"order": order,
"user": user,
"product": product,
"logistics": logistics
})
After：asyncio并发重构
# app_async.py - 重构后
import asyncio
import httpx
from flask import Flask, jsonify
app = Flask(__name__)
# 复用连接池，全局单例
client = httpx.AsyncClient(timeout=1.5, limits=httpx.Limits(max_connections=50, max_keepalive_connections=20))
semaphore = asyncio.Semaphore(20) # 下游保护阈值
async def fetch_with_limit(client, url, params):
async with semaphore: # 控制并发尖峰
resp = await client.get(url, params=params)
return resp.json()
async def fetch_all(order):
# 并发执行3个下游调用
user_task = fetch_with_limit(client, f"http://user-service/api/user/{order['uid']}", {})
product_task = fetch_with_limit(client, f"http://product-service/api/product/{order['pid']}", {})
logi_task = fetch_with_limit(client, f"http://logistics-service/api/logistics/{order['oid']}", {})
user, product, logistics = await asyncio.gather(
user_task, product_task, logi_task
)
return user, product, logistics
@app.route("/api/order/")
def get_order(order_id):
order = {"oid": order_id, "uid": 123, "pid": 456} # 模拟MySQL查询
user, product, logistics = asyncio.run(fetch_all(order)) # 关键：事件循环驱动
return jsonify({
"order": order,
"user": user,
"product": product,
"logistics": logistics
})
注意：asyncio.run()每次调用会创建新的事件循环，这是可行的，但有一定开销。更好的做法是在worker启动时创建事件循环，用loop.run_until_complete()。不过在Flask同步模型中，asyncio.run()简单且够用，开销约0.3ms。
五、踩坑与优化：三个真实教训
1. 协程泄漏：忘记await导致警告
重构第一版时，我在fetch_all里写了user_task = fetch_with_limit(...)，但忘了await gather，结果接口什么都没返回，后台疯狂刷RuntimeWarning: coroutine was never awaited。这个问题在本地单测时能发现，但压测时容易忽略——压测脚本只看状态码，不看内容。
2. Semaphore初始值过大
第一次压测把Semaphore设为100，结果下游服务直接打满CPU，RT从400ms飙到800ms。后来根据下游单机承受能力（约60 QPS），反推并发数：60 * 0.4s = 24，设为20比较安全。
3. 混用同步阻塞库
有一次我在协程里不小心用了time.sleep(0.1)模拟延迟，结果整个事件循环被阻塞，并发直接退化成串行。在协程里只能用await asyncio.sleep()。同样，如果某个下游调用是同步的requests，不要放进协程里——它会阻塞事件循环。确保所有IO操作都是异步的。
六、效果数据：压测实测对比
压测环境：4C8G 云主机，gunicorn 8 workers，压测工具 wrk，时长5分钟，并发连接500。
| 指标 | 重构前(requests同步) | 重构后(httpx+asyncio) | 提升幅度 |
|---|---|---|---|
| 平均RT | 1432ms | 452ms | 3.17x |
| P99 RT | 2810ms | 683ms | 4.11x |
| QPS | 320 | 1024 | 3.2x |
| 下游服务CPU | 28% | 61% | 更充分利用 |
| 本机CPU | 35% | 68% | 更充分利用 |
补充说明：
- 交换机的带宽波动导致RT有小幅抖动，但整体趋势稳定。
- 加了uvloop后，QPS提升至1180，因为uvloop替换了默认的asyncio事件循环，减少了epoll回调开销。
- aiohttp与httpx性能几乎一致（QPS差不到2%），最终选httpx是因为它的API更现代，支持HTTP/2（上游网关支持）。
七、总结：什么时候该用asyncio？
这次重构的收益主要来自将3次串行IO变成并发IO，而不是asyncio本身。如果你的接口也是“读多个下游服务再聚合”的模式，asyncio几乎是最优解。
但要注意：
- 如果下游服务不支持并发（比如单连接串行协议），asyncio帮不了你
- 如果接口是CPU密集型（大量本地计算），asyncio反而降低性能
- 如果QPS已经很高（>2000），建议直接换FastAPI + 原生的async def，省去asyncio.run的切换开销
我的建议是：先压测，用py-spy找出瓶颈在IO等待还是CPU计算。如果是前者，果断上asyncio；如果是后者，考虑多进程或C扩展。 别为了异步而异步，我见过有人把纯计算函数也包成协程，性能反而降了30%。
最后附上完整的依赖配置，方便复现：
Flask==2.2.5
gunicorn==20.1.0
httpx==0.25.2
uvloop==0.19.0
启动命令：gunicorn -w 8 -b 0.0.0.0:5000 app_async:app -k sync
如果你们也面临类似问题，欢迎在评论区交流具体细节。下次我会写一篇关于如何用asyncio.Queue实现全异步流水线，以及如何优雅地处理超时和重试。
