一、问题背景：一个被IO拖死的聚合接口
事情是这样的，我们有个 /api/v1/dashboard 接口，逻辑很简单：
- 调用户服务拿用户信息（HTTP）
- 调订单服务拿最近订单（HTTP）
- 调统计服务拿汇总数据（HTTP）
- 把三份数据拼起来返回
每个上游平均耗时：用户服务 80ms，订单服务 220ms，统计服务 150ms。三个请求是串行的，所以单次接口耗时大概 450ms 起步。
Flask 默认是同步阻塞的，配合 gunicorn 的 sync worker，一个 worker 一次只能处理一个请求。当时线上是 4 个 worker，理论上限也就 4 / 0.45 ≈ 8.9 QPS——当然因为有些请求慢一些，实测在 100 并发下 QPS 只有 120（因为 gunicorn 配了 threads，但 GIL 下 IO 还是能切换的，所以比纯进程好一些），P99 到了 1800ms。
业务方天天催：“为什么首页加载要转三秒？” 我看了一眼代码，这就是典型的 IO 密集型场景被同步模型坑了。
二、环境与版本
先说清楚环境，避免版本差异导致行为不一致：
- Python：3.11.6（3.11 的 asyncio 性能比 3.8 好不少，尤其 TaskGroup）
- Flask：2.3.3（仅作为对比基线）
- gunicorn：21.2.0
- aiohttp：3.9.1
- FastAPI：0.104.1（重构后用的框架）
- uvicorn：0.24.0
- wrk：4.2.0（压测工具）
- 机器：4 核 8G，CentOS 7.9，内网环境
选择 FastAPI + uvicorn 而不是“Flask + asyncio”，是因为 Flask 对 async 的支持一直很别扭，FastAPI 原生就是 ASGI，配合 aiohttp 做客户端最自然。
三、方案设计
思路很直接：
- 把三个上游调用改成并发发起，用
asyncio.gather聚合 - 用
aiohttp.ClientSession复用连接，避免每次新建 TCP - 给每个上游设置独立的
timeout，防止某个慢接口拖垮整体 - 用
asyncio.Semaphore限制并发，保护上游 - 返回值聚合逻辑保持不变，保证接口契约不变
关键点：并发不是无脑并发。上游服务也有容量上限，所以我用信号量把对单个上游的并发控制在合理范围。
四、核心实现（含 before/after）
4.1 Before：Flask 同步版本
# app_sync.py
import requests
from flask import Flask, jsonify
app = Flask(__name__)
USER_URL = "http://user-svc/api/user/{}"
ORDER_URL = "http://order-svc/api/orders?uid={}"
STATS_URL = "http://stats-svc/api/summary?uid={}"
@app.route("/api/v1/dashboard/")
def dashboard(uid):
# 串行调用，每个请求都要等
user = requests.get(USER_URL.format(uid), timeout=2).json()
orders = requests.get(ORDER_URL.format(uid), timeout=2).json()
stats = requests.get(STATS_URL.format(uid), timeout=2).json()
return jsonify({
"user": user,
"orders": orders,
"stats": stats,
})
if __name__ == "__main__":
app.run(port=8000)
启动命令：
gunicorn -w 4 -k gthread --threads 8 -b 0.0.0.0:8000 app_sync:app
4.2 After：FastAPI + asyncio 版本
# app_async.py
import asyncio
import aiohttp
from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
USER_URL = "http://user-svc/api/user/{}"
ORDER_URL = "http://order-svc/api/orders?uid={}"
STATS_URL = "http://stats-svc/api/summary?uid={}"
# 全局 session，复用连接池
session: aiohttp.ClientSession | None = None
# 对每个上游单独限流，保护下游
USER_SEM = asyncio.Semaphore(50)
ORDER_SEM = asyncio.Semaphore(50)
STATS_SEM = asyncio.Semaphore(50)
@app.on_event("startup")
async def startup():
global session
connector = aiohttp.TCPConnector(
limit=200, # 总连接上限
limit_per_host=100, # 单 host 上限
ttl_dns_cache=300, # DNS 缓存 5 分钟
keepalive_timeout=30,
)
timeout = aiohttp.ClientTimeout(total=2, connect=0.5, sock_read=1.5)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
@app.on_event("shutdown")
async def shutdown():
await session.close()
async def fetch(sem, url):
async with sem:
async with session.get(url) as resp:
resp.raise_for_status()
return await resp.json()
@app.get("/api/v1/dashboard/{uid}")
async def dashboard(uid: int):
# 三个上游并发发起
user_task = asyncio.create_task(fetch(USER_SEM, USER_URL.format(uid)))
order_task = asyncio.create_task(fetch(ORDER_SEM, ORDER_URL.format(uid)))
stats_task = asyncio.create_task(fetch(STATS_SEM, STATS_URL.format(uid)))
user, orders, stats = await asyncio.gather(
user_task, order_task, stats_task, return_exceptions=True
)
# 部分失败降级，不直接 500
if isinstance(user, Exception):
user = {"error": "user_svc_unavailable"}
if isinstance(orders, Exception):
orders = {"error": "order_svc_unavailable"}
if isinstance(stats, Exception):
stats = {"error": "stats_svc_unavailable"}
return JSONResponse({"user": user, "orders": orders, "stats": stats})
启动命令：
uvicorn app_async:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop --http httptools
注意这里用了 uvloop 和 httptools，实测在 3.11 上比默认的 asyncio 事件循环和 h11 解析器吞吐高 15%~25%。
4.3 压测脚本
# 100 并发，持续 30 秒
wrk -t4 -c100 -d30s --latency http://127.0.0.1:8000/api/v1/dashboard/12345
五、踩坑与优化
这一节是重点，因为我确实踩了不少坑。
坑 1：全局 ClientSession 必须在事件循环里创建。
一开始我在模块级别直接 session = aiohttp.ClientSession()，结果启动就报 Timeout context manager should be used inside a task。原因是 ClientSession 绑定事件循环，模块导入时还没有 running loop。改成在 startup 事件里创建就没事了。
坑 2：timeout 要分层次。
aiohttp.ClientTimeout(total=2) 只给总超时，但连接超时和读超时要分开设。我一开始只设 total，结果某个上游 TCP 握手就卡了 1.9s，几乎把 total 用光。改成 connect=0.5, sock_read=1.5 后，连接慢的情况能快速失败。
坑 3：Semaphore 不是越多越好。
我一开始给每个上游设了 200 并发信号量，结果上游被打爆，反而 P99 飙升。后来压测调到 50，QPS 反而更高——上游稳定了，整体就快了。这个值要根据上游容量来，不是拍脑袋。
坑 4：不要在 async 函数里写阻塞代码。
我一开始在 dashboard 里顺手写了个 json.dumps 处理大对象，还有一段用 requests 调内部埋点的代码，结果事件循环直接被阻塞。后来 json.dumps 换成 orjson.dumps，requests 换成 aiohttp，或者用 asyncio.to_thread 丢到线程池。
坑 5：uvicorn 的 workers 和 asyncio 的关系。
--workers 4 起的是 4 个进程，每个进程有自己的事件循环。这跟 gunicorn 多 worker 是一个道理，目的是利用多核。别以为 asyncio 单进程就能打满 4 核，CPU 密集部分还是得靠多进程。
六、效果数据
压测条件统一：wrk，4 线程，100 并发，30 秒，同一台机器，同一组上游 mock 服务（延迟固定：80/220/150ms）。
| 版本 | QPS | P50 | P90 | P99 | 错误率 |
|---|---|---|---|---|---|
| Flask + gunicorn (4w×8t) | 120 | 780ms | 1420ms | 1800ms | 0% |
| FastAPI + asyncio (4 workers) | 2100 | 45ms | 110ms | 180ms | 0% |
| FastAPI + asyncio (uvloop) | 2480 | 38ms | 92ms | 150ms | 0% |
QPS 提升约 17.5 倍，P99 降低约 90%。这不是理论值，是实测数据。原因也简单：原来 4 个 worker 最多同时处理 32 个请求（每个请求阻塞 450ms），现在每个 worker 的事件循环能同时挂起上千个协程，IO 等待期间 CPU 去处理别的请求了。
内存方面：Flask 版本 4 个 worker 常驻约 280MB，FastAPI 版本 4 个 worker 常驻约 210MB，反而更低，因为协程比线程轻量得多。
七、总结
这次重构让我再次确认一个事：IO 密集型服务，同步模型天生吃亏。不是 Flask 不好，是它在 IO 等待上没优势。asyncio + aiohttp + FastAPI 这套组合，在聚合类接口上基本是降维打击。
几点建议给准备迁移的同学：
- 先确认你的瓶颈在 IO 而不是 CPU，否则 asyncio 帮不了你
- 全局 ClientSession 一定要在事件循环里创建和关闭
- timeout 分层设置，connect / read / total 都要管
- 信号量限流保护上游，值要通过压测确定
- 别在协程里写阻塞代码，
to_thread是你的朋友 - uvloop + httptools 能白捡 15% 性能，生产环境建议开
最后，如果你的老系统没法整体迁到 FastAPI，也可以只把慢接口用 asyncio 重写，通过 nginx 分流，逐步替换。我们就是这么干的，风险可控，收益也拿到了。
