一、问题背景
先交代下场景。我们有个商品详情页的聚合接口 GET /api/v1/product/{id}，它本身不存业务数据，做的事情是把散落在各处的数据拼起来返回给前端：
- 调商品中心 HTTP 接口拿基础信息（约 80ms）
- 调库存中心 HTTP 接口拿实时库存（约 60ms）
- 调营销中心 HTTP 接口拿优惠券信息（约 100ms）
- 查本地 MySQL 拿扩展属性（约 30ms）
同步版本用的是 Flask + requests，代码大概长这样：
# before: app.py (Flask 2.2.5 + requests 2.31.0)
import requests
from flask import Flask, jsonify
from sqlalchemy import create_engine, text
app = Flask(__name__)
engine = create_engine("mysql+pymysql://user:pwd@127.0.0.1:3306/shop", pool_size=10)
@app.route("/api/v1/product/")
def product_detail(pid):
base = requests.get(f"http://product-svc/product/{pid}", timeout=2).json()
stock = requests.get(f"http://stock-svc/stock/{pid}", timeout=2).json()
promo = requests.get(f"http://promo-svc/coupon/{pid}", timeout=2).json()
with engine.connect() as conn:
rows = conn.execute(text("SELECT k,v FROM product_ext WHERE pid=:p"), {"p": pid}).fetchall()
return jsonify({"base": base, "stock": stock, "promo": promo, "ext": dict(rows)})
四个依赖串行执行，理论耗时 80+60+100+30 = 270ms，加上框架开销和网络抖动，实测平均 350ms 左右。压测（wrk，4 线程 100 连接）结果：QPS 120，P99 1.2s。机器是 4核8G，CPU 利用率只有 15%，全在等 IO。这就是典型的 IO 密集型场景，同步模型下线程池根本喂不饱。
二、环境与版本
改造前先把版本钉死，避免踩到旧版本的坑：
- Python 3.11.6（3.11 的 asyncio 在性能上有明显优化，比 3.10 快 10%~20%）
- FastAPI 0.109.2 + uvicorn 0.27.0（标准 extra，即
uvicorn[standard]，带 uvloop） - aiohttp 3.9.3（HTTP 客户端）
- asyncpg 0.29.0（MySQL 场景其实该用 aiomysql，我这里扩展属性库恰好是 PG，正好用 asyncpg）
- uvloop 0.19.0（uvicorn[standard] 会自动装）
- 压测工具：wrk 4.2.0
部署：4核8G 容器，uvicorn 启动参数 --workers 4 --loop uvloop --http httptools，单 worker 单进程。
三、方案设计
核心思路就三条：
1. 把串行变并行。 四个依赖之间没有数据依赖，全部丢进 asyncio.gather 并发执行，理论耗时从 270ms 降到 max(80,60,100,30) ≈ 100ms。
2. 连接池复用。 aiohttp 的 TCPConnector 设置 limit=100、limit_per_host=50，asyncpg 用 create_pool(min_size=5, max_size=20)。这里的关键是在应用启动时创建、生命周期内复用，而不是每次请求新建——后者是新手最常犯的错，性能会比同步还差。
3. 超时与降级。 每个下游调用单独设 timeout（ClientTimeout(total=1.5)），gather 用 return_exceptions=True，某个下游挂了就返回降级数据，不让整个接口 500。
架构上没有引入消息队列或缓存，纯粹是执行模型的改造，改动范围可控。
四、核心实现
先看改造后的完整代码：
# after: main.py (FastAPI 0.109.2 + aiohttp 3.9.3 + asyncpg 0.29.0)
import asyncio
import asyncpg
import aiohttp
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
state = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# 全局连接池，应用生命周期内复用
state["http"] = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=1.5, connect=0.5),
connector=aiohttp.TCPConnector(limit=100, limit_per_host=50, ttl_dns_cache=300),
)
state["pg"] = await asyncpg.create_pool(
"postgresql://user:pwd@127.0.0.1:5432/shop",
min_size=5, max_size=20, command_timeout=1.0,
)
yield
await state["http"].close()
await state["pg"].close()
app = FastAPI(lifespan=lifespan)
async def fetch_json(session, url, fallback):
try:
async with session.get(url) as resp:
if resp.status != 200:
return fallback
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError):
return fallback # 降级，不抛异常
async def fetch_ext(pool, pid):
try:
async with pool.acquire() as conn:
rows = await conn.fetch("SELECT k, v FROM product_ext WHERE pid=$1", pid)
return {r["k"]: r["v"] for r in rows}
except Exception:
return {}
@app.get("/api/v1/product/{pid}")
async def product_detail(pid: int):
http, pg = state["http"], state["pg"]
base, stock, promo, ext = await asyncio.gather(
fetch_json(http, f"http://product-svc/product/{pid}", {}),
fetch_json(http, f"http://stock-svc/stock/{pid}", {"stock": -1}),
fetch_json(http, f"http://promo-svc/coupon/{pid}", {"coupons": []}),
fetch_ext(pg, pid),
return_exceptions=False, # fetch_* 内部已兜底，这里不需要再吞
)
if not base:
raise HTTPException(status_code=503, detail="product service unavailable")
return {"base": base, "stock": stock, "promo": promo, "ext": ext}
几个设计点值得展开说：
fetch_json内部 try/except 兜底，gather 用默认的return_exceptions=False。 这样异常在子协程内部就被消化了，gather 拿到的都是正常值，逻辑更清晰。如果反过来让 gather 吞异常，那每个返回值都得判断是不是 Exception 实例，代码很脏。TCPConnector的ttl_dns_cache=300很重要。默认 DNS 缓存时间短，高 QPS 下 DNS 解析会成为隐藏瓶颈，我实测这一项让 P99 降了约 15ms。ClientTimeout(total=1.5, connect=0.5)分开设。连接超时短，快速失败；总超时留足下游处理时间。只设 total 的话，某个下游 TCP 握手卡住会白白占满 1.5s。
五、踩坑与优化
改造过程中踩了三个印象深刻的坑，记录一下。
坑 1：在协程里调了同步库，事件循环被阻塞。
第一版我图省事，扩展属性那块直接用了原来的 SQLAlchemy 同步 engine，想着"就 30ms 无所谓"。结果压测 QPS 死活上不去 400。原因很明确：同步 DB 调用会阻塞整个事件循环，单 worker 下所有并发请求都被这一个 30ms 卡住，并发度直接归零。后来换成 asyncpg，QPS 立刻翻了几倍。教训：异步代码里不能有任何同步 IO，包括日志的文件写入、time.sleep、requests。
坑 2：连接池配小了，高并发下请求排队。
TCPConnector 一开始设的 limit=20，压测到 800 QPS 时大量请求报 Connection pool is full。单个 worker 同时处理几百个请求，每个请求要 3 个 HTTP 连接，20 的上限显然不够。调到 limit=100 后正常。经验值：limit ≈ 预期并发数 × 下游调用数 × 1.2。asyncpg 的 max_size=20 也是同理，从 5 调到 20 后 DB 等待时间从 40ms 降到 8ms。
坑 3：uvicorn 多 worker 下的连接池是每进程独立的。
我一开始以为 4 个 worker 共享一个池，按总并发算的池大小，结果每个进程都开一份，总连接数是配置值的 4 倍，把下游服务的连接数打满了。后来按单进程的并发量重新算的池大小。这个点文档里写得不显眼，但线上很容易翻车。
另外顺手做的一个优化：把 uvicorn 的 --http httptools 和 --loop uvloop 显式打开（uvicorn[standard] 装了就默认启用，但显式写出来更保险），压测对比 uvloop 比默认的 asyncio 事件循环 QPS 高约 8%。
六、效果数据
压测条件保持一致：wrk 4 线程 100 连接，持续 60s，4核8G 容器，下游服务是 mock 的固定延迟（80/60/100ms）。
| 指标 | 同步版 (Flask+requests) | 异步版 (FastAPI+asyncio) | 提升 |
|---|---|---|---|
| QPS | 120 | 1900 | 15.8x |
| 平均延迟 | 350ms | 52ms | -85% |
| P99 延迟 | 1200ms | 180ms | -85% |
| CPU 利用率 | 15% | 68% | - |
| 内存占用 | 210MB | 180MB | -14% |
几个观察：
- QPS 提升主要来自并发化（串行 270ms → 并行 100ms，理论 2.7x）和事件循环的高并发承载能力（单进程可同时挂几千个协程），两者叠加远不止 2.7x。
- P99 从 1.2s 降到 180ms，主要归功于超时隔离——同步版一个下游抖动会拖垮整个请求，异步版 1.5s 直接降级返回。
- 异步版不是没有代价：调试难度上升，异常栈会跨协程，出问题时排查更费劲。所以我在
fetch_json里对非 200 状态做了日志埋点，方便定位是哪个下游降级了。
七、总结
这次改造的核心结论：IO 密集型 Web API 在同步模型下 CPU 吃不满、线程全在等，asyncio 是对症的解法，但收益取决于三个细节——是否真的全链路异步、连接池是否按并发量配够、超时降级是否到位。 三个里任何一个没做对，QPS 都上不去。
适用边界也说清楚：如果接口里有 CPU 密集计算（比如图片处理、复杂加解密），asyncio 帮不上忙，得上多进程或 run_in_executor；如果下游依赖本身有强顺序依赖，并发化也发挥不出来。我们这个场景四个依赖完全独立，属于 asyncio 的甜点区。
最后提醒一句：异步改造别一把梭，先把慢接口的依赖关系画清楚，确认没有隐藏的同步调用（包括日志、配置中心 SDK、监控埋点这些容易被忽略的地方），再动手。我踩的坑 1 就是栽在一个"就 30ms 无所谓"的同步 DB 调用上。
