一、问题背景
去年接手了一个电商项目的商品详情聚合接口。这个接口的作用很简单：前端传一个商品ID，后端把商品基础信息、库存、价格、评价、推荐位这5块数据聚合起来返回。每块数据来自不同的下游服务，都是HTTP调用。
最初的实现是最朴素的同步写法，用requests逐个调用。单看代码逻辑很清晰，但压测结果很难看：
- 4核8G机器，50并发压测，QPS 120左右
- P99响应时间1.8秒，P50也有600ms
- CPU利用率不到15%，大量时间耗在等IO
问题很明确：5个下游调用串行执行，假设每个平均耗时100ms，光串行等待就是500ms。这500ms里线程完全阻塞，什么也干不了。加机器能缓解，但成本不划算，而且接口延迟本身也没降下来。
当时考虑过几个方案：线程池并发、Go重写、asyncio。线程池改动最小但线程开销和上下文切换在50并发下也不便宜；Go重写要动整个服务，不现实。最后选了asyncio，因为服务本身就是Python，下游全是IO密集的HTTP调用，正好是asyncio的主场。
二、环境与版本
先把环境交代清楚，asyncio在不同Python版本差异不小，尤其是3.8之后的一些变化：
- Python 3.11.4（从3.8升级上来的，3.11的asyncio性能有明显优化）
- aiohttp 3.9.1
- FastAPI 0.109.0 + uvicorn 0.25.0（Web框架本来就在用FastAPI，天然支持async）
- requests 2.31.0（同步版本用）
- 压测工具：wrk 4.2.0，50并发连接，持续60秒
机器配置：4核8G，下游服务是内网调用，网络延迟稳定在1-2ms。
有一点要提前说：FastAPI的同步路由函数默认跑在线程池里，异步路由函数跑在事件循环里。这个区别很关键，后面踩坑部分会细说。
三、方案设计
改造思路不复杂，核心就三点：
1. 串行改并行。 5个下游调用彼此没有依赖，完全可以并发发出，总耗时从「5个之和」变成「最慢的那个」。
2. 用aiohttp替代requests。 requests是同步阻塞的，在事件循环里直接调用会卡死整个循环。aiohttp的ClientSession是异步的，天然适配。
3. 连接池复用。 每次请求新建TCP连接开销不小，用ClientSession维护连接池，限制总连接数和单host连接数。
架构上保持简单，不引入额外的任务队列或消息中间件，就是接口内部把5个调用并发化。超时控制用asyncio.wait_for包一层，单个下游超时不影响整体，返回降级数据。
这里有个设计决策值得说：5个下游里，商品基础信息是必须的，拿不到就直接报错；库存、价格、评价、推荐位这4个可以降级。所以并发的时候用asyncio.gather配合return_exceptions=True，基础信息单独处理，其余4个失败就返回默认值。
四、核心实现
Before：同步串行版本
先看改造前的代码，这是最典型的同步写法：
# before_sync.py
import requests
from fastapi import FastAPI, HTTPException
app = FastAPI()
BASE = "http://internal.api"
TIMEOUT = 3
@app.get("/product/{pid}")
def get_product_detail(pid: int):
# 5个下游调用，串行执行
try:
base = requests.get(f"{BASE}/base/{pid}", timeout=TIMEOUT).json()
except Exception:
raise HTTPException(status_code=502, detail="base service error")
try:
stock = requests.get(f"{BASE}/stock/{pid}", timeout=TIMEOUT).json()
except Exception:
stock = {"count": 0}
try:
price = requests.get(f"{BASE}/price/{pid}", timeout=TIMEOUT).json()
except Exception:
price = {"amount": 0}
try:
review = requests.get(f"{BASE}/review/{pid}", timeout=TIMEOUT).json()
except Exception:
review = {"score": 0, "count": 0}
try:
recommend = requests.get(f"{BASE}/recommend/{pid}", timeout=TIMEOUT).json()
except Exception:
recommend = []
return {
"base": base,
"stock": stock,
"price": price,
"review": review,
"recommend": recommend,
}
这段代码逻辑上没毛病，但5个requests.get是串行的。压测时P50 600ms，基本就是5个下游耗时之和。
After：asyncio并发版本
改造后的代码，用aiohttp + asyncio.gather：
# after_async.py
import asyncio
import aiohttp
from fastapi import FastAPI, HTTPException
app = FastAPI()
BASE = "http://internal.api"
TIMEOUT = 3.0
CONNECTOR_LIMIT = 100 # 连接池总上限
CONNECTOR_LIMIT_PER_HOST = 50 # 单host上限
# 全局session，应用生命周期内复用
session: aiohttp.ClientSession = None
@app.on_event("startup")
async def startup():
global session
connector = aiohttp.TCPConnector(
limit=CONNECTOR_LIMIT,
limit_per_host=CONNECTOR_LIMIT_PER_HOST,
ttl_dns_cache=300, # DNS缓存5分钟
enable_cleanup_closed=True,
)
timeout = aiohttp.ClientTimeout(total=TIMEOUT, connect=1.0)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
@app.on_event("shutdown")
async def shutdown():
await session.close()
async def fetch(path: str, default=None, required=False):
"""单个下游调用，带超时和异常处理"""
try:
async with session.get(f"{BASE}{path}") as resp:
if resp.status != 200:
if required:
raise HTTPException(status_code=502, detail=f"{path} status {resp.status}")
return default
return await resp.json()
except asyncio.TimeoutError:
if required:
raise HTTPException(status_code=504, detail=f"{path} timeout")
return default
except aiohttp.ClientError:
if required:
raise HTTPException(status_code=502, detail=f"{path} client error")
return default
@app.get("/product/{pid}")
async def get_product_detail(pid: int):
# 5个调用并发发出
base_task = fetch(f"/base/{pid}", required=True)
stock_task = fetch(f"/stock/{pid}", default={"count": 0})
price_task = fetch(f"/price/{pid}", default={"amount": 0})
review_task = fetch(f"/review/{pid}", default={"score": 0, "count": 0})
recommend_task = fetch(f"/recommend/{pid}", default=[])
base, stock, price, review, recommend = await asyncio.gather(
base_task, stock_task, price_task, review_task, recommend_task,
return_exceptions=False, # base失败会直接抛出，其余已内部降级
)
return {
"base": base,
"stock": stock,
"price": price,
"review": review,
"recommend": recommend,
}
几个关键点说明：
全局session。 aiohttp的ClientSession不该每次请求新建，应该在应用启动时创建、关闭时销毁。每次新建session会重新建连接池，等于把连接复用的好处全丢了。这一点和requests的Session用法一致，但很多人写aiohttp时容易忽略。
连接池参数。 limit=100是总连接上限，limit_per_host=50是单host上限。下游虽然是同一个域名，但走的是内网，压测时50并发连接够用。这两个值要根据实际并发量和下游承载能力调，设太小会排队，设太大可能压垮下游。
超时分层。 ClientTimeout里total=3秒是总超时，connect=1秒是建连超时。建连超时单独设短一点，因为内网建连正常就几毫秒，1秒还没连上基本是下游挂了，没必要等满3秒。
降级逻辑。 fetch函数的default参数配合required标记，required=True的（基础信息）失败直接抛异常，其余返回默认值。这样gather里不需要return_exceptions=True，逻辑更清晰。
五、踩坑与优化
改造过程不是一次成功的，有几个坑值得记录。
坑1：在async路由里调用了同步阻塞代码。
第一版改完，QPS只从120涨到300，远低于预期。查了半天发现，降级逻辑里有个地方调用了同步的日志写入（写本地文件），这个操作在事件循环里阻塞了。事件循环是单线程的，任何一个阻塞调用都会卡住所有协程。后来把日志改成异步队列写入，QPS直接跳到1500+。
这个坑很典型：asyncio的并发是「协作式」的，不是抢占式。代码里只要有一处同步阻塞（文件IO、CPU密集计算、同步HTTP），整个循环都会被拖住。排查方法可以用asyncio的debug模式，设置PYTHONASYNCIODEBUG=1，慢回调会打警告。
坑2：连接池太小导致排队。
一开始connector的limit设了20，压测时QPS卡在800上不去，P99反而升高。原因是50并发下，每个请求要占5个连接（5个下游调用），20个连接根本不够，大量请求在等连接池释放。调到100后QPS到2100。这个值不是越大越好，要结合下游能承受的并发看，我们下游单机能扛200并发，100的总连接上限是安全的。
坑3：uvicorn的worker数。
uvicorn默认单worker，也就是单进程单事件循环。4核机器只用了1核。改成--workers 4后，每个worker独立的事件循环，QPS从2100涨到接近8000（线性扩展）。这里要注意，多worker下全局session是每个进程各一份，连接池也是各算各的，所以limit参数要按单worker的并发来算，不能按总量算。
优化：DNS缓存。
aiohttp默认会缓存DNS，但ttl_dns_cache默认值是0，等于不缓存。内网虽然是IP直连影响不大，但如果是域名调用，每次请求都解析DNS会有额外开销。设成300秒后，域名场景下P50能降10ms左右。
优化：响应体预读。
async with session.get(...)里，resp.json()是流式读取的。如果下游返回体较大，可以在拿到resp后先await resp.read()一次性读完再解析，减少协程切换次数。我们的返回体都不大（几KB），这个优化收益不明显，但大响应体场景下有用。
六、效果数据
压测条件统一：wrk 50并发连接，持续60秒，4核8G，下游服务内网延迟1-2ms。
| 指标 | Before（同步串行） | After（asyncio并发） | 提升 |
|---|---|---|---|
| QPS | 120 | 2100（单worker） | 17.5x |
| P50 | 600ms | 45ms | 13x |
| P99 | 1800ms | 210ms | 8.6x |
| CPU利用率 | 12% | 68% | - |
| 内存占用 | 380MB | 420MB | 略增 |
多worker（4 worker）下QPS约7900，P99 230ms。CPU打满到95%左右，说明单机已经压到瓶颈，再往上要加机器或优化下游。
P99 210ms比P50 45ms高不少，主要是少数请求命中了某个下游的慢响应（下游P99本身有150ms左右），并发化之后整体耗时取决于最慢的那个调用，这也是并发化的固有特性——木桶效应从「求和」变成了「取最大」。
内存略增是因为连接池和协程对象，420MB对8G机器来说完全可接受。
七、总结
这次改造的核心结论：
-
IO密集的聚合接口，串行改并行的收益是数量级的。 5个100ms的调用，串行500ms，并行100ms，理论5倍，实测因为下游有快有慢，整体提升更明显。
-
asyncio的坑主要在「阻塞」二字。 事件循环里任何同步阻塞都会拖垮全局，改造时要地毯式排查同步IO、CPU计算、同步锁。debug模式是排查利器。
-
连接池参数要结合并发模型算。 单worker和多worker下连接池是独立的，limit要按单进程并发量估，不能拍脑袋。
-
不是所有接口都值得改。 纯CPU密集或下游调用少的接口，asyncio收益有限，反而增加代码复杂度。这次改造针对的是5个独立下游调用的聚合场景，收益明确。
如果重来一次，我会在改造前先加一个下游调用的耗时监控，把每个下游的P50/P99打点，这样并发化之后能快速定位是哪个下游拖了后腿。这次是改造完才补的监控，走了点弯路。
代码已经上线跑了半年多，稳定性没问题。后续如果要进一步优化，方向是把高频商品的数据加本地缓存，减少下游调用次数，那是另一个话题了。
