一、问题背景：一个"看起来没问题"的接口
去年接了个聚合查询接口，业务很简单：前端传一个 userId，后端要去三个下游服务拿数据——用户基础信息、订单统计、积分余额，拼起来返回。
最初用 Flask 写的，代码大概长这样：
# app_sync.py
import requests
from flask import Flask, jsonify, request
app = Flask(__name__)
BASE = "http://downstream.internal"
def fetch_user(uid):
return requests.get(f"{BASE}/user/{uid}", timeout=2).json()
def fetch_orders(uid):
return requests.get(f"{BASE}/orders/{uid}", timeout=2).json()
def fetch_points(uid):
return requests.get(f"{BASE}/points/{uid}", timeout=2).json()
@app.route("/profile/")
def profile(uid):
user = fetch_user(uid)
orders = fetch_orders(uid)
points = fetch_points(uid)
return jsonify({"user": user, "orders": orders, "points": points})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, threaded=True)
逻辑上完全正确，单看每个下游接口的响应时间也不高——平均 40ms 左右。但上线后监控报警：P99 稳定在 1.2 秒，高峰期接口超时率 3%。
原因很直白：三个请求是串行的。40ms × 3 = 120ms 起步，再加上 Flask 默认线程池的调度开销和下游偶发抖动，P99 冲到秒级毫不意外。而压测数据更难看——单进程 QPS 只有 120 左右。
问题的本质不是"代码写得烂"，而是IO 密集场景下用了同步阻塞模型。每个请求都在等网络，CPU 全程闲着。
二、环境与版本
先说清楚环境，避免你照着抄跑不起来：
- Python 3.11.6（3.10+ 都行，3.11 的 asyncio 性能有优化）
- Flask 3.0.0
- aiohttp 3.9.1
- uvloop 0.19.0（Linux 下必装，性能提升明显）
- 压测工具：wrk 4.2.0
- 系统：Ubuntu 22.04，4 核 8G，下游服务在同一内网
关键点：aiohttp 3.9 和 Python 3.11 搭配最稳，早期 aiohttp 3.8 在 3.11 上有过 SSL 相关的兼容问题，别踩。
三、方案设计：为什么是 asyncio + aiohttp
有几个候选方案，我简单对比下思路：
- 多线程 + requests：能并发，但线程切换开销大，1000 并发就要 1000 个线程，内存扛不住。
- gevent monkey patch：能用，但 monkey patch 是全局副作用，和现有代码混用容易出玄学 bug。
- asyncio + aiohttp：单线程事件循环，IO 等待时切走，内存开销小，代码可控。
我选 3。设计要点：
- 用
aiohttp.ClientSession复用连接，不要每个请求新建 session（这是最大的性能坑） - 三个下游调用用
asyncio.gather并发 - 用
uvloop替换默认事件循环 - 用
asyncio.wait_for给每个下游加独立超时 - Flask 本身是同步的，用
asgiref的async_to_sync桥接，或者干脆换 FastAPI。我这里为了改动最小，用 asgiref 桥接。
四、核心实现
4.1 改造后的代码
# app_async.py
import asyncio
import aiohttp
from asgiref.sync import async_to_sync
from flask import Flask, jsonify
app = Flask(__name__)
BASE = "http://downstream.internal"
# 全局 session，进程启动时创建一次
_session = None
async def get_session():
global _session
if _session is None or _session.closed:
connector = aiohttp.TCPConnector(
limit=200, # 总连接数上限
limit_per_host=100, # 单 host 上限
ttl_dns_cache=300, # DNS 缓存 5 分钟
keepalive_timeout=30,
)
_session = aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=2),
)
return _session
async def fetch(session, path):
async with session.get(f"{BASE}{path}") as resp:
resp.raise_for_status()
return await resp.json()
async def fetch_all(uid):
session = await get_session()
# 并发发起，gather 默认 return_exceptions=False，任一失败即抛
user, orders, points = await asyncio.gather(
fetch(session, f"/user/{uid}"),
fetch(session, f"/orders/{uid}"),
fetch(session, f"/points/{uid}"),
)
return {"user": user, "orders": orders, "points": points}
@app.route("/profile/")
def profile(uid):
data = async_to_sync(fetch_all)(uid)
return jsonify(data)
4.2 启动时挂上 uvloop
# run.py
import asyncio
import uvloop
uvloop.install() # 必须在事件循环创建前调用
from app_async import app
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, threaded=True)
uvloop.install() 全局替换事件循环策略，实测在 IO 密集场景下比默认的 selector 事件循环快 2~3 倍。注意它只支持 Linux/macOS，Windows 上会直接报错，用 sys.platform 判断一下更稳。
五、踩坑与优化
坑 1：gather 的异常处理
asyncio.gather 默认 return_exceptions=False，意思是任意一个协程抛异常，整个 gather 立即失败，其他协程会被取消但不会有任何提示。这在聚合接口里很危险——积分服务挂了，用户信息也拿不到。
我的处理是拆开：
async def fetch_safe(session, path, default=None):
try:
return await fetch(session, path)
except Exception as e:
# 这里最好打日志，别静默吞掉
return default
async def fetch_all(uid):
session = await get_session()
user, orders, points = await asyncio.gather(
fetch_safe(session, f"/user/{uid}"),
fetch_safe(session, f"/orders/{uid}", default={"points": 0}),
fetch_safe(session, f"/points/{uid}", default={"points": 0}),
)
return {"user": user, "orders": orders, "points": points}
用户信息是核心，失败就返回 None 让上层报错；订单和积分降级返回默认值。这样部分失败不影响整体可用性。
坑 2：连接池配置不当反而更慢
我第一版没配 TCPConnector，用默认值。默认 limit=100，但 limit_per_host=0 表示不限。看起来更宽松，实际上压测时出现大量 TIME_WAIT，连接反复重建。
调整后：
connector = aiohttp.TCPConnector(
limit=200,
limit_per_host=100, # 关键：限制单 host，避免打爆下游
ttl_dns_cache=300,
keepalive_timeout=30, # 保持长连接
)
keepalive_timeout=30 让连接在空闲 30 秒内复用，避免每次请求都三次握手。这一项就把 P99 从 80ms 压到了 50ms 以下。
优化 3：超时分层
外层 ClientTimeout(total=2) 是兜底，但下游每个接口的合理超时不同。用户信息 200ms 就够，订单统计可能要 500ms。用 asyncio.wait_for 单独控制：
user = await asyncio.wait_for(fetch(session, f"/user/{uid}"), timeout=0.3)
超时设置太统一，会导致快接口被慢接口拖累。
六、效果数据
压测命令：
wrk -t4 -c200 -d30s --latency http://127.0.0.1:8000/profile/12345
对比数据（下游三接口平均响应各 40ms）：
| 指标 | Flask 同步版 | asyncio 版 | 提升 |
|---|---|---|---|
| QPS | 120 | 2300 | 19.2x |
| P50 | 820ms | 42ms | 19.5x |
| P99 | 1200ms | 48ms | 25x |
| 内存峰值 | 380MB | 145MB | 降低 62% |
内存下降是因为不用维护线程池，事件循环单线程处理所有连接。
一个细节：QPS 2300 是单进程数据。如果上 gunicorn 多 worker，注意uvloop 要装在每个 worker 里，且每个 worker 有独立的事件循环和 session，别指望共享。
七、总结
这次重构的核心就三件事：
- 识别 IO 密集：串行调用多个下游，就是 asyncio 的主场。CPU 密集别用它，GIL 会让你怀疑人生。
- 连接池比语法重要：
gather谁都会写，但TCPConnector的limit_per_host和keepalive_timeout才是性能分水岭。 - 异常要显式处理：
gather的默认行为在聚合场景下是反直觉的，return_exceptions=True或者拆开 try 都行，别让一个下游拖垮全局。
如果你的接口也是"查多个下游拼结果"这种模式，且 P99 一直下不去，先别急着加机器，把它改成 asyncio 试试。改动量不大，收益很直接。
有踩过类似坑的欢迎评论区聊聊，尤其是 aiohttp 在极端并发下的表现，我这边 5000 并发时遇到过 Too many open files，后来调了 ulimit -n 到 65535 才稳住，这块有空再单独写一篇。
