一、问题背景：一个被同步IO拖死的商品详情API
去年接手了一个商品详情聚合接口，逻辑不复杂：客户端传商品ID，服务端要去三个下游拿数据——商品基础信息（MySQL）、库存（内部HTTP服务）、促销标签（另一个HTTP服务），最后拼装返回。
最早的实现是Flask + requests，代码写得很直白：
# before: app_sync.py
import requests
from flask import Flask, jsonify
from sqlalchemy import create_engine
app = Flask(__name__)
engine = create_engine("mysql+pymysql://user:pwd@db:3306/shop", pool_size=20)
@app.route("/api/item/")
def item_detail(item_id):
with engine.connect() as conn:
row = conn.execute(
"SELECT id, name, price FROM item WHERE id=%s", (item_id,)
).fetchone()
stock = requests.get(f"http://stock-svc/stock/{item_id}", timeout=2).json()
promo = requests.get(f"http://promo-svc/promo/{item_id}", timeout=2).json()
return jsonify({
"id": row[0], "name": row[1], "price": float(row[2]),
"stock": stock["count"], "promo": promo["tags"],
})
这个接口的问题在于：三个IO操作是完全串行的，MySQL 30ms + 库存服务 40ms + 促销服务 50ms，理论最快也要120ms。但实际压测（wrk -t8 -c200）下来，QPS只有120，P99延迟1.8s。
原因很清楚：Flask默认用gunicorn的sync worker，每个worker同时只处理一个请求，而requests.get是阻塞调用，一个请求在等下游的时候，这个worker就干等着。线程池一旦被打满，请求就开始排队，延迟雪崩。
我试过最粗暴的方案：gunicorn开32个worker。QPS确实上去了，到了900左右，但内存占用翻了4倍，而且下游服务稍微抖一下（比如促销服务P99到500ms），worker又全部堵死。这是典型的"用资源换并发"，治标不治本。
二、环境与版本
这次改造用的环境：
- Python 3.11.6（选3.11是因为asyncio在3.11有显著的性能优化，Task创建开销比3.10低约20%）
- aiohttp 3.9.1
- asyncmy 0.2.9（异步MySQL驱动，比aiomysql快，底层是Cython实现）
- FastAPI 0.109.0 + uvicorn 0.25.0（ASGI框架，替代Flask）
- gunicorn 21.2.0（只用来做进程管理，worker class用uvicorn.workers.UvicornWorker）
选aiohttp而不是httpx，是因为这个场景只需要简单的GET，aiohttp的连接池控制和超时配置更细，而且在我实测里单核吞吐略高。httpx的API更友好，如果你们的项目已经用了httpx，换过去也没问题。
三、方案设计：并发编排而不是串行等待
核心思路就一句话：三个下游调用互相不依赖，为什么要串行等？
用asyncio.gather并发发起三个请求，总耗时从"三者之和"变成"三者最大值"。理论延迟从120ms降到50ms左右。
但光gather还不够，有几个设计点必须考虑：
- 并发上限控制。如果每个请求都无脑发3个下游请求，QPS 2000时就是每秒6000个下游调用，下游服务会被打挂。用
asyncio.Semaphore限制同时在途的下游请求数。 - 超时要分级。单个下游超时设2s，但整个请求的总超时设3s，用
asyncio.wait_for包住，防止某个下游hang住导致请求堆积。 - 连接池复用。aiohttp的
TCPConnector必须配置limit和limit_per_host，否则默认100的总连接数在高并发下会成为瓶颈。 - 降级策略。库存和促销不是核心数据，拿不到就返回默认值，不能让整个接口500。
四、核心实现
先看改造后的完整代码：
# after: app_async.py
import asyncio
import asyncmy
import aiohttp
from fastapi import FastAPI
from contextlib import asynccontextmanager
app = FastAPI()
# 全局连接池，在lifespan里初始化
pool: asyncmy.Pool = None
session: aiohttp.ClientSession = None
# 限制同时在途的下游请求，防止打挂下游
DOWNSTREAM_SEM = asyncio.Semaphore(200)
@asynccontextmanager
async def lifespan(app: FastAPI):
global pool, session
pool = await asyncmy.create_pool(
host="db", user="user", password="pwd", db="shop",
minsize=10, maxsize=50,
)
connector = aiohttp.TCPConnector(
limit=500, # 总连接数上限
limit_per_host=200, # 单host上限
ttl_dns_cache=300,
enable_cleanup_closed=True,
)
timeout = aiohttp.ClientTimeout(total=2, connect=0.5)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
yield
await session.close()
pool.close()
await pool.wait_closed()
app.router.lifespan_context = lifespan
async def fetch_json(url: str, default):
"""带信号量控制的下游请求，失败降级"""
async with DOWNSTREAM_SEM:
try:
async with session.get(url) as resp:
if resp.status != 200:
return default
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError):
return default
async def get_item_from_db(item_id: int):
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute(
"SELECT id, name, price FROM item WHERE id=%s", (item_id,)
)
return await cur.fetchone()
@app.get("/api/item/{item_id}")
async def item_detail(item_id: int):
async def _detail():
return await get_item_from_db(item_id)
# 三个调用并发发起，库存/促销失败降级
detail, stock, promo = await asyncio.gather(
_detail(),
fetch_json(f"http://stock-svc/stock/{item_id}", {"count": -1}),
fetch_json(f"http://promo-svc/promo/{item_id}", {"tags": []}),
return_exceptions=False,
)
if detail is None:
return {"code": 404, "msg": "item not found"}
return {
"id": detail[0], "name": detail[1], "price": float(detail[2]),
"stock": stock["count"], "promo": promo["tags"],
}
几个关键点展开说：
并发编排：asyncio.gather把三个协程同时调度，事件循环在等待MySQL返回时去处理HTTP响应，IO等待时间被完全重叠。注意return_exceptions=False，因为我在fetch_json内部已经做了降级，不需要gather层再兜底。
Semaphore的位置：信号量只包住下游HTTP调用，不包DB查询。因为DB连接池本身有maxsize=50做限流，再加一层信号量是重复控制。200这个值是按"下游服务能承受的QPS / 单机QPS"估算的——库存服务压测能扛8000 QPS，单机目标2000 QPS × 1个库存请求 = 2000，留4倍余量取200有点保守，实际调到500也没问题，但初期保守点更安全。
超时分层：ClientTimeout(total=2, connect=0.5)，连接超时0.5s快速失败，总超时2s。外层还有uvicorn的--timeout-keep-alive 5。
启动命令：
gunicorn app_async:app \
-k uvicorn.workers.UvicornWorker \
-w 4 \
--bind 0.0.0.0:8000 \
--timeout 10 \
--max-requests 10000 \
--max-requests-jitter 1000
4个worker（机器是4核8G），每个worker一个独立事件循环。--max-requests防止内存泄漏累积，加了jitter避免4个worker同时重启。
五、踩坑与优化
坑1：在async函数里调同步库，整个事件循环卡死
改造初期我在item_detail里留了一行日志用的同步requests.get调监控服务，压测时QPS死活上不去，只有400。用py-spy dump了一下发现事件循环卡在socket recv上——同步IO把整个worker的循环堵死了，一个请求阻塞，这个worker上的其他几百个协程全部停摆。
教训：async函数里绝对不能出现任何阻塞调用。DB驱动、HTTP客户端、文件IO、甚至time.sleep都要换成异步版本。排查手段就是py-spy或者asyncio.debug模式（PYTHONASYNCIODEBUG=1会打印执行超过100ms的回调）。
坑2：连接池没配，高并发下大量TIME_WAIT
一开始TCPConnector用的默认配置（limit=100），QPS到1500左右开始出现Connection pool is full警告，延迟抖动。把limit调到500、limit_per_host调到200后解决。另外记得开enable_cleanup_closed=True，某些下游服务会主动关连接，不开这个会泄漏。
坑3：DB连接池在fork时被复制
gunicorn用preload模式时，如果在模块顶层创建连接池，4个worker会共享同一批连接，导致协议错乱。正确做法是把池的创建放在lifespan（每个worker启动时各自建池），我上面的代码就是这么写的。
优化：给热点商品加了一层进程内缓存
压测发现商品基础信息查询占了DB压力的70%，而商品详情是典型的读多写少。加了个cachetools.TTLCache(maxsize=10000, ttl=30)，命中率约65%，DB QPS直接降了一半。注意TTLCache是同步的，但纯内存操作耗时在微秒级，不构成阻塞，可以放心在协程里用。
六、效果数据
压测条件：wrk -t8 -c200 -d60s，同一台4核8G机器，下游服务mock了固定延迟（库存40ms、促销50ms、DB 30ms）。
| 指标 | 改造前（Flask+requests, 8 worker） | 改造后（FastAPI+asyncio, 4 worker） |
|---|---|---|
| QPS | 120 | 2100 |
| P50 延迟 | 420ms | 38ms |
| P99 延迟 | 1800ms | 85ms |
| 单机内存 | 1.2GB | 480MB |
| 部署机器数 | 8 | 2 |
QPS提升17.5倍，P99降低95%。线上灰度一周后，高峰期CPU使用率从85%降到42%，下游服务也没再出现被打挂的情况（Semaphore限流起了作用）。
需要说明的是，2100 QPS是在mock下游、固定延迟的理想条件下测的。真实环境下游有长尾，上线后实测稳定QPS在1600左右，P99约120ms，依然比改造前好一个数量级。
七、总结
这次改造最大的收获不是QPS数字，而是想清楚了一件事：同步框架下的并发瓶颈，加机器和加worker只能缓解，不能解决。IO密集型服务的本质是"等待"，asyncio把等待时间重叠起来，才是对症下药。
给准备做类似改造的同学几条建议：
- 先profile再动手，确认瓶颈真的是IO等待，别把CPU密集型服务硬套asyncio。
- 信号量和超时一定要配，异步不是无限并发，失控的并发比串行更可怕。
- 改造期间用py-spy定期dump，同步调用混进事件循环是最高频的事故来源。
- 降级逻辑在异步代码里要写得更小心，
gather里一个协程抛异常会影响整体行为，return_exceptions的取值要想清楚。
代码已脱敏，逻辑可直接跑。有问题的欢迎评论区交流。
