1. 问题背景：一个看似无辜的聚合接口
事情起因是上周一早上，运维发来一张监控截图：我们的/api/v1/user-summary接口P95耗时2.1秒，而SLA要求是1.5秒。这个接口的逻辑很简单——它需要调用三个内部服务：
- 用户基础信息服务（平均响应 800ms）
- 用户订单统计服务（平均响应 600ms）
- 用户推荐位服务（平均响应 700ms）
我打开代码一看，血压直接拉满：
# app/routes.py (改造前 - 串行调用)
def get_user_summary(user_id: str):
user_info = requests.get(f"http://user-svc/users/{user_id}").json()
order_stats = requests.get(f"http://order-svc/users/{user_id}/orders/stats").json()
rec_items = requests.get(f"http://rec-svc/users/{user_id}/recommendations").json()
return {
"user": user_info,
"orders": order_stats,
"recommendations": rec_items
}
三个独立的HTTP请求，没有任何数据依赖，却老老实实地排队执行——总耗时800+600+700=2100ms。这就像去银行办事，明明有三个窗口，你却只在第一个窗口排完队再去第二个窗口。
2. 环境与版本：为什么选httpx而不是aiohttp
先交代环境：
- Python 3.10.8（关键：3.10+的asyncio对task处理有性能优化）
- Flask 2.2.3（同步框架，通过asyncio.run桥接）
- httpx 0.23.3（支持HTTP/2，连接池复用比aiohttp更成熟）
- 压测工具：wrk 4.2.0，单机4核8G
有人可能问：为什么不用aiohttp？我承认aiohttp性能很强，但httpx的API兼容Requests，迁移成本极低——requests.get(...)改成httpx.AsyncClient.get(...)，基本只需要改装饰器。而且httpx支持HTTP/2多路复用，对微服务网关场景有额外收益。
3. 方案设计：用asyncio.gather把串行变并发
核心思路很简单：用asyncio.gather()把三个IO操作并发出去，总耗时约等于最慢的那个请求（800ms），而不是三个之和。
但有几个关键决策：
- 如何创建事件循环：Flask是同步框架，不能直接
await。我选择在视图函数内用asyncio.run()创建独立事件循环。注意：每次请求创建新loop会有开销（约1-2ms），但换来的是线程安全——因为Flask开发模式下默认多线程，共享loop会出大问题。 - 连接池复用：必须用
httpx.AsyncClient作为上下文管理器，或者显式close。如果每次请求都新建client，TCP握手和TLS开销会吃掉大部分收益。 - 异常隔离：三个服务任何一个挂掉，
gather默认会取消其他任务。我设置return_exceptions=True，让失败的服务返回None，保证接口不整体崩溃。
4. 核心实现：改造后的代码
这是改造后的完整代码块，注意几个细节：
# app/routes.py (改造后 - 异步并发)
import asyncio
import httpx
from flask import jsonify
# 全局复用AsyncClient，但注意线程安全问题
_client = None
def get_client():
global _client
if _client is None:
_client = httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=100))
return _client
async def fetch_json(client, url: str):
try:
resp = await client.get(url)
resp.raise_for_status()
return resp.json()
except Exception as e:
# 记录日志，返回None让上层处理降级
app.logger.error(f"fetch {url} failed: {e}")
return None
async def async_summary(user_id: str):
client = get_client()
# 关键：gather并发，return_exceptions让单个失败不影响整体
user_info, order_stats, rec_items = await asyncio.gather(
fetch_json(client, f"http://user-svc/users/{user_id}"),
fetch_json(client, f"http://order-svc/users/{user_id}/orders/stats"),
fetch_json(client, f"http://rec-svc/users/{user_id}/recommendations"),
return_exceptions=True
)
return {
"user": user_info,
"orders": order_stats,
"recommendations": rec_items
}
# 同步入口：asyncio.run桥接
def get_user_summary_sync(user_id: str):
return asyncio.run(async_summary(user_id))
# Flask路由保持不变
@app.route('/api/v1/user-summary/')
def user_summary(user_id):
result = get_user_summary_sync(user_id)
return jsonify(result)
三个容易踩的坑：
- 全局Client被多线程竞争：Flask默认threaded=True，多个请求线程共享
_client。理论上httpx的AsyncClient是线程安全的（内部有锁），但我实测在高并发下偶尔出现Event loop is closed错误。解决方案：改用threading.local()存储client，或者干脆每个线程创建自己的loop。 - asyncio.run不能嵌套：如果你的Flask app本身跑在事件循环里（比如用Quart），就不能用asyncio.run。这里我们是纯同步Flask，所以没问题。
- DNS解析阻塞：httpx默认的解析是同步的，会阻塞事件循环。我加了
trust_env=True和系统DNS缓存，问题不大。如果追求极致，可以用anyio的后端。
5. 踩坑与优化：我踩过的三个大坑
坑1：gather的返回顺序问题。gather不保证返回顺序和输入顺序一致！如果你的聚合结果有顺序依赖，必须用asyncio.wait或手动保持映射。我这里三个结果互相独立，所以无所谓。
坑2：超时和重试策略。上游服务偶尔会抖动，我设置了timeout=10.0（比原来requests默认的30s更激进），同时加了重试装饰器：
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=0.5, max=3))
async def fetch_json(client, url):
resp = await client.get(url)
resp.raise_for_status()
return resp.json()
注意：重试必须放在fetch_json内部，而不是gather外层，否则会重复整个并发批次。
坑3：事件循环泄漏。最初版本在视图函数里写loop = asyncio.new_event_loop()，然后手动loop.run_until_complete()，最后没关loop——导致内存涨到2G。后来统一用asyncio.run()，它内部会正确关闭loop。但如果你的代码在Jupyter或某些环境下，asyncio.run可能报错，这时需要手动loop.close()。
6. 效果数据：从2.1s到0.31s
用wrk压测，4线程，200连接，持续60秒：
| 指标 | 改造前(串行) | 改造后(异步) | 提升倍数 |
|---|---|---|---|
| 平均响应 | 2.1s | 0.31s | 6.8x |
| P95 | 2.3s | 0.35s | 6.6x |
| QPS | 120 | 980 | 8.2x |
| 错误率 | 0.1% | 0.05% | - |
注意：0.31s约等于三个服务中最慢的那个（800ms）不可能，因为我的测试环境里三个mock服务的响应时间被我降低了（因为压测机带宽瓶颈）。真实环境里，如果三个服务都是800ms，那么异步总耗时约820ms（800ms + 20ms调度开销），仍然比串行的2.4s快3倍。
额外收获：CPU占用率从改造前的78%降到21%——因为原来每个请求都阻塞在requests库的socket等待上，现在事件循环把等待时间让给了其他任务。
7. 总结与建议
这次改造让我对asyncio有了更务实的认识：
- asyncio不是银弹：只对IO密集型有效。如果逻辑里有CPU计算（比如JSON解析大数组），需要用
loop.run_in_executor混合使用。 - 连接池是隐形MVP：我一开始没复用一个client，压测QPS只有500。复用后直接翻倍到980。
- 监控要跟上：异步代码的trace和日志链路会变复杂，建议用
contextvars传递request_id。
最后给个实用建议：如果你的Flask项目要大规模异步化，不如直接迁移到FastAPI或Quart——它们原生支持async路由，不需要asyncio.run桥接，性能更稳。但如果是存量Flask项目，用本文这套方案是性价比最高的。
有任何问题欢迎评论区交流，特别是遇到Event loop is closed或RuntimeError: asyncio.run() cannot be called from a running event loop的朋友，我保证你排完坑会回来感谢我的。
