一、问题背景：一个“看起来很轻”的API为什么这么慢
事情要从去年Q4说起。我们有个内部商品查询API，逻辑简单到不能再简单：接收一批SKU（平均20个），逐个调用上游商品中心的HTTP接口拿详情，聚合后返回。日均请求200万，峰值QPS约300。
代码是典型的Flask同步写法，核心就一个循环：
# before: app.py (Flask 2.2.5 + requests 2.31.0)
@app.route("/api/products", methods=["POST"])
def get_products():
skus = request.json["skus"]
results = []
for sku in skus:
resp = requests.get(f"{UPSTREAM}/product/{sku}", timeout=3)
results.append(resp.json())
return {"data": results}
单看没毛病，但压测一跑就露馅：单实例QPS只有120，P99 1.8s。原因很直白——每个请求串行发20次HTTP，每次上游RT约8ms，光网络等待就160ms，而这160ms里worker线程啥也干不了，纯阻塞。Gunicorn开了4个sync worker，等于最多4个请求并行，QPS自然上不去。高峰期8台机器CPU利用率不到15%，全在等IO。
结论很明确：这是典型的IO密集型场景，线程模型是错的，该上asyncio。
二、环境与版本
先交代下环境，避免版本差异导致行为不一致：
- Python 3.11.6（3.11的asyncio有task group和更快的调度器，比3.8快一截）
- aiohttp 3.9.1（客户端用，别用3.8.x，连接池有已知bug）
- Flask 2.2.5 → 改造后保留Flask做路由，用
asgiref桥接，或者直接换FastAPI。我选了后者，见下文 - FastAPI 0.109.0 + uvicorn 0.25.0（
--workers 4，--loop uvloop） - uvloop 0.19.0（Linux下事件循环提速明显，实测调度开销降约30%）
- 压测工具：wrk 4.2.0，
wrk -t4 -c200 -d30s
上游商品中心接口不变，RT均值8ms，P99 25ms，单机限流2000 QPS。
三、方案设计
改造思路分三层：
1. 框架层：Flask的同步WSGI模型没法直接跑async view（硬跑会报coroutine was never awaited），要么用asgiref.sync.async_to_sync桥接，要么换ASGI框架。我直接换FastAPI，原生async，顺便白嫖Pydantic做入参校验。
2. 并发层：20个SKU的请求用asyncio.gather并发发出，总耗时从20 × 8ms压到接近1 × 8ms。但要注意上游限流，不能无脑并发，加asyncio.Semaphore控制单请求并发数（我设的10）。
3. 连接层：aiohttp的TCPConnector必须复用连接，否则每次新建TCP+TLS握手，省下的时间全还回去。limit设100，limit_per_host设50，keepalive_timeout设30s。
还有个细节：上游是HTTP/1.1，单个连接同一时刻只能跑一个请求，所以limit_per_host不能太小，否则并发被连接数卡死。这是我们压测时发现的，后面踩坑部分细说。
四、核心实现
4.1 after 代码
# after: main.py (FastAPI 0.109.0 + aiohttp 3.9.1)
import asyncio
import aiohttp
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
UPSTREAM = "http://product-center.internal"
SEM = asyncio.Semaphore(10) # 单请求内最大并发
session: aiohttp.ClientSession | None = None
class QueryReq(BaseModel):
skus: list[str]
@app.on_event("startup")
async def startup():
global session
connector = aiohttp.TCPConnector(
limit=100, # 全局连接池上限
limit_per_host=50, # 单host上限，必须 >= SEM
keepalive_timeout=30,
enable_cleanup_closed=True,
)
timeout = aiohttp.ClientTimeout(total=3, connect=0.5)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
@app.on_event("shutdown")
async def shutdown():
await session.close()
async def fetch_one(sku: str) -> dict:
async with SEM:
async with session.get(f"{UPSTREAM}/product/{sku}") as resp:
resp.raise_for_status()
return await resp.json()
@app.post("/api/products")
async def get_products(req: QueryReq):
results = await asyncio.gather(
*(fetch_one(s) for s in req.skus),
return_exceptions=True, # 单个失败不拖垮整批
)
data = [r for r in results if not isinstance(r, Exception)]
return {"data": data, "failed": len(results) - len(data)}
启动命令：
uvicorn main:app --host 0.0.0.0 --port 8000 \
--workers 4 --loop uvloop --http httptools
几个关键点解释一下：
return_exceptions=True：商品中心偶尔对下架SKU返回404，不加这个参数整个gather直接抛异常，一批全废。加了之后失败项单独丢弃，接口返回failed计数。Semaphore(10)：不是拍脑袋。上游限流2000 QPS，我们单实例峰值约1100 QPS × 平均20 SKU = 22000次上游调用/秒？不对，这里算错了——是1100 QPS的请求，每个请求内部20个SKU并发，但Semaphore限的是单请求内并发，所以单请求最多10个上游调用同时在飞。总上游QPS = 1100 × 20 / (20/10的并发批次数)……实际压测时上游QPS稳定在1800左右，没触发限流。这个值要根据上游配额调，别照抄。connect=0.5：连接超时单独设短，避免上游半死不活时连接阶段就卡满3秒。
4.2 一个容易忽略的点：session的生命周期
ClientSession必须在事件循环内创建，且全局复用。我见过有人在每个请求里async with aiohttp.ClientSession()，那连接池等于没有，性能比同步还差。放在startup里创建一次，shutdown里关闭，是标准做法。
五、踩坑与优化
坑1：RuntimeError: Event loop is closed
第一版把session创建在模块顶层（session = aiohttp.ClientSession()），uvicorn多worker下直接炸。原因是模块导入时还没有运行中的事件循环，aiohttp绑定到了错误或已关闭的loop。改成startup钩子里创建就好了。这个错在FastAPI + aiohttp组合里非常常见，搜一下一堆人踩。
坑2：连接池打满导致延迟飙升
最初limit_per_host设的20，压测到800 QPS时P99突然从200ms跳到2s。抓包发现大量请求在等连接。原因前面提过：HTTP/1.1单连接串行，20个连接最多支撑20个并发请求，而我们单请求内Semaphore是10、4个worker → 理论并发40，连接不够用。把limit_per_host提到50后恢复正常。经验公式：limit_per_host >= workers × SEM，再留点余量。
坑3：信号量跨请求泄漏
早期版本把SEM定义在handler内部（每次请求新建），看着没问题，但配合return_exceptions=True时，如果某个fetch_one在async with SEM之前就被取消（客户端断开连接），信号量不会释放——其实新建的Semaphore不受影响，真正的问题是全局Semaphore + 客户端提前断开时，gather被取消，部分task卡在async with SEM里，取消传播不及时会短暂占住槽位。解决办法是确保取消能正常传播，别在fetch_one里吞CancelledError。这个坑比较隐蔽，压测时表现为偶发的吞吐抖动。
优化：uvloop + httptools
换--loop uvloop --http httptools后，同等压力下CPU sys时间降了约25%，QPS从950提到1100。纯Python的默认loop在高并发短连接场景下调度开销确实明显。
六、效果数据
wrk压测，同样的请求体（20 SKU），4 worker，同规格机器（4C8G）：
| 指标 | before (Flask+requests) | after (FastAPI+aiohttp) | 提升 |
|---|---|---|---|
| QPS | 120 | 1100 | 9.2x |
| P50 延迟 | 420ms | 85ms | 4.9x |
| P99 延迟 | 1800ms | 210ms | 8.6x |
| 单实例CPU | 12% | 68% | - |
| 上游调用QPS | 2400 | 1800 | 降25%* |
| 生产机器数 | 8 | 2 | -75% |
*上游QPS反而降了，因为连接复用减少了重复的鉴权/握手开销，且失败重试少了。
线上跑了两个月，日均200万请求，P99稳定在200-250ms，2台机器CPU峰值75%，还有余量。唯一注意的是内存比之前高（连接池+协程对象），单实例从300MB涨到550MB，可以接受。
七、总结
这次改造的核心不是“用了asyncio”，而是认清了瓶颈在IO等待。同步模型下加机器是线性堆成本，异步模型下把等待时间重叠起来，单机就能吃下原来8台的量。几个可复用的经验：
ClientSession全局单例，startup建、shutdown关，别在请求里新建。limit_per_host必须大于等于并发度，HTTP/1.1下连接数就是并发上限。asyncio.gather配return_exceptions=True，部分失败不影响整体。- 上游有限流就一定加
Semaphore，别让并发数失控。 - 生产环境上uvloop，白捡的20%+性能。
最后提醒一句：asyncio不是银弹。如果链路里有CPU密集操作（比如大批量JSON解析、图片处理），该扔线程池扔线程池（loop.run_in_executor），别在事件循环里硬算，否则一个慢操作卡死所有协程。我们这批SKU的JSON都很小（平均2KB），所以没这个问题，但换个大payload场景就得重新评估了。
