1. 问题背景：一个“看似正常”的慢接口
先交代业务场景：内部的一个订单详情聚合API，前端一次调用，后端需要并行拉取用户服务、订单服务、物流服务的数据，然后做字段拼接返回。代码逻辑很简单，最初版本是串行requests调用：
# 伪代码：串行调用上游
def get_order_detail(order_id):
user = requests.get(f"http://user-service/{order_id}/user", timeout=0.5).json()
order = requests.get(f"http://order-service/{order_id}/order", timeout=0.5).json()
logistic = requests.get(f"http://logistic-service/{order_id}/ship", timeout=0.5).json()
return merge(user, order, logistic)
线上表现：每个上游服务P99约200ms，串行加起来600ms，加上网络抖动和超时重试，接口P99直接到2.8秒。压测单机QPS 320，CPU利用率不到15%——典型的I/O密集型瓶颈，线程池能缓解但Python的GIL和线程切换开销浪费了大量CPU。
2. 环境与版本：Python 3.11 + 关键依赖
先说版本，不同的asyncio行为差异很大。我的生产环境：
- 操作系统：Ubuntu 22.04 LTS
- Python：3.11.7（3.10+才有
asyncio.timeout，3.11的TaskGroup更稳定） - Web框架：Flask 3.0.0（仅作为API入口，不改造其自身）
- HTTP客户端：httpx 0.26.0（比aiohttp更贴近requests的API）
- 压测工具：wrk 4.2.0（本机回环测试，注意tcp_nodelay）
关键点：Flask是同步WSGI框架，不能直接在view里await。所以方案是：保持Flask外壳，内部用asyncio.run()驱动一个新的异步调度层。这也是老项目渐进式改造的常见手法，不用重写整个服务。
3. 方案设计：用asyncio.gather替代串行阻塞
核心思路：将三次上游HTTP调用改为并发协程。我们写了一个独立的order_fetcher.py，封装并发逻辑，Flask view只负责同步调用asyncio.run。
设计要点：
- 使用httpx.AsyncClient复用连接池（limits参数调大，默认10太小，压测会报ConnectionPoolFull）。
- 超时用asyncio.timeout（3.11新增），比wait_for更优雅，支持asyncio.CancelledError的干净处理。
- 并发粒度：三个协程通过asyncio.gather(return_exceptions=True)包裹，即使一个上游挂了，其他两个结果也能返回，并记录错误日志。
4. 核心实现：Before/After代码对比
Before（同步阻塞版）
# app_sync.py
import requests, time
from flask import Flask, jsonify
app = Flask(__name__)
def fetch_url(url, timeout=0.5):
try:
return requests.get(url, timeout=timeout).json()
except Exception as e:
return {"error": str(e)}
@app.route('/order/')
def order_detail(order_id):
start = time.perf_counter()
user = fetch_url(f"http://user-service/{order_id}/user")
order = fetch_url(f"http://order-service/{order_id}/order")
logistic = fetch_url(f"http://logistic-service/{order_id}/ship")
result = {"user": user, "order": order, "logistic": logistic}
result["elapsed_ms"] = (time.perf_counter() - start) * 1000
return jsonify(result)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, threaded=True)
After（asyncio并发版）
# app_async.py
import asyncio, httpx, time
from flask import Flask, jsonify
app = Flask(__name__)
# 连接池：最多100个连接，超过后等待而不是报错
limits = httpx.Limits(max_connections=100, max_keepalive_connections=20)
client = httpx.AsyncClient(limits=limits, timeout=httpx.Timeout(0.6))
async def fetch_json(client, url):
try:
resp = await client.get(url)
return resp.json()
except Exception as e:
return {"error": str(e)}
async def fetch_all(order_id):
urls = [
f"http://user-service/{order_id}/user",
f"http://order-service/{order_id}/order",
f"http://logistic-service/{order_id}/ship"
]
# gather并发，return_exceptions保证部分失败不阻塞
results = await asyncio.gather(
*(fetch_json(client, u) for u in urls),
return_exceptions=True
)
return results
@app.route('/order/')
def order_detail(order_id):
start = time.perf_counter()
# 关键：asyncio.run创建独立事件循环，避免与Flask的线程冲突
user, order, logistic = asyncio.run(fetch_all(order_id))
result = {"user": user, "order": order, "logistic": logistic}
result["elapsed_ms"] = (time.perf_counter() - start) * 1000
return jsonify(result)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, threaded=True)
注意：asyncio.run每次调用会创建和销毁事件循环，开销约0.1ms，可忽略。但如果有全局连接池，必须在进程启动时创建一次，不能在协程内重复创建AsyncClient——否则连接池无法复用。
5. 踩坑与优化：三个真实教训
坑1：事件循环嵌套问题。最初我在协程内直接调用asyncio.run，导致RuntimeError: asyncio.run() cannot be called from a running event loop。因为httpx的AsyncClient内部会自动创建事件循环（在异步上下文外）。解决方案：将asyncio.run放在Flask view里（同步函数），不要嵌套在协程内。
坑2：连接池耗尽导致超时。压测时发现QPS到800左右就出现大量ReadTimeout，日志显示ConnectionPoolFull。这是因为httpx默认max_connections=10，10个并发请求就打满了。调参后稳定：max_connections=100，max_keepalive_connections=20。同时建议设置http2=True（需要安装httpx[http2]），HTTP/2多路复用能进一步减少TCP握手。
坑3：协程泄漏与未取消任务。如果客户端断开连接，Flask的请求线程会被终止，但asyncio.run正在运行的协程不会自动取消，导致协程泄漏（内存缓慢增长）。优化：用asyncio.timeout包裹整个fetch_all，超时后主动取消所有子任务：
async def fetch_all_with_timeout(order_id, timeout=0.6):
try:
async with asyncio.timeout(timeout):
return await fetch_all(order_id)
except asyncio.TimeoutError:
# 超时后自动取消子协程，返回错误占位符
return [{"error": "timeout"} for _ in range(3)]
6. 效果数据：压测对比
用wrk压测本机（回环地址，避免网络干扰），模拟10个并发连接，持续30秒：
| 指标 | Before（同步requests） | After（asyncio+httpx） | 提升幅度 |
|---|---|---|---|
| QPS（请求/秒） | 320 | 1890 | 5.9倍 |
| 平均延迟（ms） | 312 | 53 | 5.9倍 |
| P99延迟（ms） | 2840 | 312 | 9.1倍 |
| CPU使用率 | 12% | 28% | 合理增长 |
| 内存占用（峰值） | 120MB | 165MB | 连接池开销 |
压测命令：wrk -t4 -c10 -d30s --latency http://127.0.0.1:5000/order/12345
数据解读：QPS提升主要来自三个上游请求从串行变并行，理论最大提升3倍，但实际5.9倍是因为httpx的keep-alive连接复用了TCP握手（requests默认每次都新建连接），以及asyncio减少了线程切换开销。P99改善明显，因为之前的串行模式下，一旦某个上游偶发500ms延迟，总延迟线性叠加；并发模式下，总延迟约等于最慢的那个上游（约200ms）。
注意：这里模拟上游服务是本地mock（固定延迟200ms），如果真实上游延迟波动更大，并发收益会更夸张。
7. 总结：适用边界与进一步优化
这个方案适合“同步框架内部做I/O并发”的场景，本质是用协程替代线程池。但要注意：
- CPU密集型任务不适合：asyncio无法利用多核，计算密集代码该用多进程（如ProcessPoolExecutor）。
- 如果Flask本身是瓶颈：需要换异步框架（FastAPI或Quart），但改造成本高。我这里QPS瓶颈在上游I/O，Flask同步框架足够。
- 进一步的优化方向：将
asyncio.run改为持久事件循环（用loop.run_in_executor），可以省去每次创建循环的开销；或者使用uvloop替换默认事件循环，实测再提升15%左右（但需要兼容性测试）。 - 监控建议：在每个协程内记录耗时和状态，用
contextvars传递请求ID，方便在日志中追踪一条完整请求的所有子调用。
最后说一句：异步编程不是银弹，但针对I/O密集型的Web API，asyncio是Python生态里性价比最高的优化手段。关键是把阻塞调用找出来，用gather或TaskGroup并行化，然后调好连接池和超时参数。代码就那些，但性能差距是数量级的。如果你们也有类似的老接口，不妨试试这个方案，压测数据会说话。
