asyncio重构同步Web API：QPS从120提升到1100的调优记录

一、问题背景：一个被串行IO拖垮的聚合接口

去年接手了一个商品详情页的后端接口 /api/v1/product/{id}。业务逻辑很简单：拿到商品ID后，依次调用三个内部服务——

• 库存服务 stock-svc：查可售库存
• 价格服务 price-svc：查实时价格和促销信息
• 评论服务 review-svc：查评论数和好评率

三个服务各自响应时间都在 150~250ms 之间，但代码是串行调用的，所以单次请求的总耗时基本就是三者之和，P99 稳定在 800ms 以上。压测（wrk，4 线程 200 连接，30 秒）下来 QPS 只有 120 左右，CPU 使用率却低得可怜——不到 15%。

问题很清楚：这是一个典型的 IO 密集型场景，线程大部分时间都在等网络，串行调用把等待时间叠加了三倍。当时的实现用的是 requests，跑在 gunicorn + 同步 Flask worker 上，worker 数量开到 8 也救不了，因为每个 worker 在等待下游响应时是完全阻塞的。

改造目标很明确：三个下游调用并发执行，用 asyncio + aiohttp 重写这个接口，同时保证超时、降级、异常处理这些生产环境必须有的东西不能丢。

二、环境与版本

先把版本钉死，异步生态里版本差异带来的坑不少：

• Python 3.11.7（3.11 的 asyncio 在 TaskGroup 和异常处理上有明显改进，建议至少 3.10+）
• aiohttp 3.9.3
• FastAPI 0.109.2 + uvicorn 0.29.0（标准 [standard] extras，带 uvloop）
• uvloop 0.19.0

• 压测工具：wrk 4.2.0，机器 4C8G，下游三个服务部署在同机房，RTT ")
  def product_detail(pid):
  # 三次串行 HTTP 调用，总耗时 = 三者之和
  stock = requests.get(STOCK_URL.format(id=pid), timeout=1.0).json()
  price = requests.get(PRICE_URL.format(id=pid), timeout=1.0).json()
  review = requests.get(REVIEW_URL.format(id=pid), timeout=1.0).json()

  return jsonify({
  "id": pid,
  "stock": stock.get("available", 0),
  "price": price.get("final_price"),
  "promotion": price.get("promotion"),
  "review_count": review.get("count", 0),
  "good_rate": review.get("good_rate", 0),
  })

### 改造后（asyncio 并发版）

```python
# after: FastAPI + aiohttp，并发调用 + 超时 + 降级
import asyncio
import aiohttp
from fastapi import FastAPI
from contextlib import asynccontextmanager

STOCK_URL = "http://stock-svc/api/stock/{id}"
PRICE_URL = "http://price-svc/api/price/{id}"
REVIEW_URL = "http://review-svc/api/review/{id}"

# 全局连接池，进程内复用
session: aiohttp.ClientSession = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global session
    connector = aiohttp.TCPConnector(
        limit=200,            # 连接池总上限
        limit_per_host=50,    # 单 host 上限
        ttl_dns_cache=300,    # DNS 缓存 5 分钟
        enable_cleanup_closed=True,
    )
    timeout = aiohttp.ClientTimeout(total=0.8, connect=0.3)
    session = aiohttp.ClientSession(connector=connector, timeout=timeout)
    yield
    await session.close()

app = FastAPI(lifespan=lifespan)

async def fetch_json(url: str, default: dict) -> dict:
    """单次下游调用，失败返回默认值，不抛异常"""
    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

@app.get("/api/v1/product/{pid}")
async def product_detail(pid: int):
    # 三个调用并发执行，整体超时 0.8s
    stock_task = fetch_json(STOCK_URL.format(id=pid), {"available": 0})
    price_task = fetch_json(PRICE_URL.format(id=pid), {"final_price": None})
    review_task = fetch_json(REVIEW_URL.format(id=pid), {"count": 0, "good_rate": 0})

    try:
        stock, price, review = await asyncio.wait_for(
            asyncio.gather(stock_task, price_task, review_task),
            timeout=0.8,
        )
    except asyncio.TimeoutError:
        # 整体超时兜底，返回部分可用数据（这里简化为空结构）
        stock, price, review = {"available": 0}, {"final_price": None}, {"count": 0}

    return {
        "id": pid,
        "stock": stock.get("available", 0),
        "price": price.get("final_price"),
        "promotion": price.get("promotion"),
        "review_count": review.get("count", 0),
        "good_rate": review.get("good_rate", 0),
    }

几个关键点解释一下：

• TCPConnector 参数：limit=200 是进程内总连接上限，limit_per_host=50 防止单个下游被打爆。这两个值要结合压测调，一开始我设的 100/20，QPS 上到 800 就开始出现连接等待。
• ClientTimeout(total=0.8, connect=0.3)：总超时 0.8s 比业务 SLA 略紧，留出余量给序列化和网络回程。connect 超时单独设 0.3s，避免 TCP 握手阶段卡太久。
• return_exceptions 的取舍：这里没用 gather(return_exceptions=True)，而是每个 task 内部自己 try/except 吞掉异常返回默认值。这样 gather 本身不会因为异常提前结束，语义更清晰。
• asyncio.wait_for 兜底：即使单个请求内部超时没触发，整体也有硬超时，防止极端情况下 gather 挂住。

五、踩坑与优化

改造过程不是一次成功的，记录几个真实踩过的坑：

坑 1：Session 不能每次请求新建。 最初图省事在路由函数里 async with aiohttp.ClientSession()，结果 QPS 不升反降，连接建立开销把并发收益吃掉了。改成 lifespan 里全局单例后正常。

坑 2：uvloop 和某些 C 扩展冲突。 项目里有个同步的 protobuf 解析库，在 uvloop 下偶发段错误。最后把那段逻辑挪到 run_in_executor 的线程池里跑才稳定。如果你们项目有类似 C 扩展，建议先小流量灰度。

坑 3：workers=4 时连接池是进程级的。 每个 worker 有自己的 200 连接池，实际总连接数是 4×200。下游服务的连接数配额要按 worker 数算，我第一次没注意，把 stock-svc 的连接数打满了，触发了它的限流。

坑 4：同步 ORM 拖后腿。 商品主表信息还是用 SQLAlchemy 同步查的，虽然只有几毫秒，但在高并发下会阻塞事件循环。后来换成了 asyncpg + SQLAlchemy 2.0 的 async 支持，或者临时用 run_in_executor 包一下。事件循环里绝对不能有阻塞调用，这是铁律。

优化点：给三个下游调用加了简单的本地缓存（cachetools.TTLCache，TTL 1s），价格和库存这种秒级变化的字段命中率不高，但评论数这种变化慢的字段命中率能到 60% 以上，进一步压低了 P99。

六、效果数据

压测条件一致：wrk 4 线程 200 连接，30 秒，下游服务响应时间人为控制在 150~250ms。

QPS 从 120 到 1100，基本符合理论预期：三个 200ms 的串行调用变成并发后，单请求耗时从 600ms 降到 200ms 左右，再叠加连接池和事件循环的效率提升，3 倍理论值加上 3 倍左右的并发效率提升，量级对得上。

CPU 从 15% 涨到 62% 是好事，说明机器终于被用起来了。内存略增主要是连接池和事件循环的开销，在可接受范围。

七、总结

这次改造的核心其实不复杂：识别出 IO 密集的串行调用，用 asyncio 并发化，再补齐连接池、超时、降级三件套。 真正花时间的是踩坑和调参，尤其是连接池配额和超时阈值的设定，必须结合压测数据反复调。

几条经验给要动手的同学：

1. 别在事件循环里跑任何同步阻塞代码，包括同步 ORM、同步文件 IO、time.sleep。
2. ClientSession 全局复用，连接池参数按下游配额和 worker 数一起算。
3. 超时一定要设，而且要分层设（connect / total / 整体 wait_for），生产环境没有超时等于没有防御。
4. 并发不等于无脑 gather，单个下游失败要有降级路径，否则一个服务抖动会放大成整体雪崩。

如果你们也有类似的串行聚合接口，建议先加个耗时打点，看看下游调用占比，大概率能靠 asyncio 拿到几倍的免费性能。