一、问题背景：接口慢在“串行等待”
上个月接手了一个Flask写的报表服务，其中一个/api/v1/dashboard接口被前端反馈“转圈超过2秒”。排查后发现，该接口需要依次调用7个下游微服务（用户画像、订单统计、库存状态、物流轨迹、价格波动、推荐列表、活动标签），每个下游响应在200-400ms之间。
关键代码长这样：
# 原始代码（串行阻塞）
def get_dashboard_data(user_id):
result = {}
result['user'] = requests.get(f"http://svc-user/profile/{user_id}").json()
result['orders'] = requests.get(f"http://svc-order/stats/{user_id}").json()
result['stock'] = requests.get(f"http://svc-stock/current/{user_id}").json()
# ... 省略另外4个请求
return result
7个请求加起来的理论耗时是1.8-2.8秒，由于是同步阻塞，CPU大部分时间在等待I/O。当时线上平均耗时1.9秒，P99直接飙到2.4秒。而真正的业务逻辑（组装数据）只花了不到10ms。
二、环境与版本：Python 3.11 + asyncio生态
- Python: 3.11.4（原生支持
asyncio.TaskGroup，比gather更优雅） - Web框架: Flask 3.0.3（改造后保留Flask作为入口，内部用asyncio跑协程）
- HTTP客户端: httpx 0.27.0（支持异步且兼容requests API）
- 服务器: gunicorn + uvicorn workers（后续会解释为什么）
三、方案设计：协程并发替代线程池
最初考虑过concurrent.futures.ThreadPoolExecutor，但Python的GIL对I/O密集任务其实影响不大，线程方案可行。不过asyncio的优势在于更轻量（单线程事件循环），且能精确控制并发数（信号量）。最终采用：
- Flask视图函数内启动事件循环（
asyncio.run()） - 用
asyncio.gather()或TaskGroup并发发起7个httpx异步请求 - 通过
asyncio.Semaphore(5)限制最大并发为5，避免下游被打爆 - 设置超时和重试策略
四、核心实现：改造后的代码
import asyncio
import httpx
from flask import Flask, jsonify
app = Flask(__name__)
# 下游服务地址配置
SERVICES = {
'user': 'http://svc-user/profile/{uid}',
'orders': 'http://svc-order/stats/{uid}',
# ... 其余5个
}
async def fetch_one(client, sem, name, url):
async with sem:
try:
resp = await client.get(url, timeout=5.0)
resp.raise_for_status()
return name, resp.json()
except Exception as e:
# 降级：返回空数据而非让整个接口失败
return name, {'error': str(e), 'fallback': True}
async def fetch_all(user_id):
sem = asyncio.Semaphore(5)
timeout = httpx.Timeout(connect=2.0, read=5.0, write=2.0, pool=2.0)
async with httpx.AsyncClient(timeout=timeout, limits=httpx.Limits(max_connections=20)) as client:
tasks = []
for name, url in SERVICES.items():
url = url.format(uid=user_id)
tasks.append(fetch_one(client, sem, name, url))
# TaskGroup会在所有任务完成或取消时退出
async with asyncio.TaskGroup() as tg:
for task in tasks:
tg.create_task(task)
# 收集结果
results = {}
for task in tasks:
name, data = task.result()
results[name] = data
return results
@app.route('/api/v1/dashboard')
def dashboard():
user_id = request.args.get('uid')
if not user_id:
return jsonify({'error': 'missing uid'}), 400
# 注意：Flask是同步的，这里用run_until_complete
loop = asyncio.new_event_loop()
try:
data = loop.run_until_complete(fetch_all(user_id))
finally:
loop.close()
return jsonify(data)
为什么不用asyncio.run()？ 因为Flask的请求处理线程中可能已经有事件循环（比如调试模式），直接asyncio.run()会报RuntimeError。用new_event_loop + run_until_complete更安全。
五、踩坑与优化：三个血泪教训
1. httpx必须用AsyncClient而不是Client
一开始我图省事，直接用requests配合asyncio.to_thread()，虽然能并发，但协程切换的开销更大。换成httpx后，连接池复用效果更好，性能提升约15%。
2. 超时设置不当导致雪崩
最初没设超时，结果下游一个服务卡死，整个事件循环被阻塞。后来设置了connect=2.0, read=5.0，并对单个请求失败做降级处理（返回fallback: true），这样即使一个服务挂了，接口依然能返回其他6个服务的数据。
3. 信号量并发控制必须加
不加信号量时，7个请求同时发出，下游服务出现连接拒绝。加Semaphore(5)后，每批最多5个并发，实测下游压力减少40%，且整体耗时只增加了30ms（从0.28s到0.31s），完全可接受。
部署优化：
Flask自带的Werkzeug服务器是同步的，无法发挥asyncio优势。最终用gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app部署，每个worker内可以跑事件循环。如果单纯用Flask+asyncio，实际上还是单线程交替执行，但I/O等待时间被压缩了。
六、效果数据：真实压测对比
用wrk -t8 -c200 -d30s压测，对比改造前后：
| 指标 | 改造前（同步） | 改造后（asyncio） | 提升 |
|---|---|---|---|
| 平均响应时间 | 1.92s | 0.31s | -83.8% |
| P99响应时间 | 2.41s | 0.42s | -82.6% |
| QPS（吞吐量） | 42 | 286 | +580% |
| 错误率 | 0.5%（超时） | 0.02%（仅个别降级） | -96% |
线上部署后，监控图表显示：接口耗时从1.9s降到0.3s左右，下游服务CPU占用率反而下降了（因为并发被限制，不再有瞬时的连接风暴）。
七、总结：asyncio是I/O密集的银弹？
这次改造让我意识到：对于I/O密集型API聚合场景，asyncio是最优解之一，但它不是万能的：
- 如果你的代码里有大量CPU计算（比如正则匹配、JSON解析大字段），需要配合
asyncio.to_thread或进程池 - 如果下游服务本身不支持高并发，信号量限流是必须的
- Flask + asyncio是权宜之策，如果从零开始，推荐FastAPI原生支持异步
最后留个思考题：如果这个接口需要处理100个下游服务，你的信号量设多少合适？我的答案是min(50, 下游服务连接数上限/2)，欢迎在评论区讨论。
（全文完）
