asyncio重构同步Web API：QPS从120提升到1800的优化过程

一、问题背景

先说结论：这不是一篇"asyncio很牛逼"的布道文，而是一次被逼出来的重构记录。

我们有一个内部服务叫 aggregator-api，职责很简单：接收一个 user_id，然后并行去 5 个下游服务拉数据（用户画像、订单摘要、积分余额、消息未读数、推荐列表），拼成一个 JSON 返回给 BFF 层。

最初的实现非常"朴素"——Flask + requests，同步串行调用：

# before: app_sync.py
import requests
from flask import Flask, jsonify

app = Flask(__name__)

DOWNSTREAMS = {
    "profile": "http://profile-svc:8001/v1/profile",
    "orders": "http://order-svc:8002/v1/summary",
    "points": "http://points-svc:8003/v1/balance",
    "messages": "http://msg-svc:8004/v1/unread",
    "recommend": "http://rec-svc:8005/v1/list",
}

@app.route("/v1/aggregate/")
def aggregate(user_id):
    result = {}
    for name, url in DOWNSTREAMS.items():
        resp = requests.get(url, params={"user_id": user_id}, timeout=3)
        result[name] = resp.json()
    return jsonify(result)

单个下游平均耗时 80~150ms，5 个串行下来就是 500~700ms。上线初期量小还能忍，等到 BFF 层开始做首页聚合、QPS 上到 100 左右时，问题彻底暴露：

• 压测（wrk，4 线程 100 连接）下 QPS 只有 120，P99 2.3s
• Flask 默认单进程，GIL + 同步阻塞 IO，请求全堵在 requests.get 上
• 下游任何一个慢，整条链路跟着慢，没有隔离

试过最省事的方案：gunicorn -w 8。QPS 涨到 400 左右就到顶了，因为每个 worker 还是同步阻塞，机器 CPU 利用率不到 15%，全在等 IO。这就是典型的 IO-bound 场景，同步模型下加进程只是线性堆资源，性价比极低。

所以决定用 asyncio 重写。

二、环境与版本

先把版本钉死，asyncio 生态版本差异挺大的，尤其是 httpx 的 0.2x 系列接口变动不少：

Python 升到 3.11 是有意的，3.11 的 asyncio 在 Task 调度和 asyncio.gather 上有明显优化（官方 benchmark 里 asyncio 相关有 10%~20% 提升），而且 TaskGroup（3.11 新增）比 gather 的错误处理更干净，后面会用到。

机器配置固定：4 核 8G，容器部署，压测机和被测服务同机房。

三、方案设计

核心思路就三条：

1. IO 全异步化：Web 层换 FastAPI（原生 ASGI），HTTP 客户端换 httpx.AsyncClient，把 5 个下游调用从串行改成 asyncio.gather 并发。
2. 连接池复用：httpx 的 AsyncClient 全局单例，配置 Limits，避免每次请求重建 TCP/TLS 连接。
3. 并发隔离：用 asyncio.Semaphore 限制单请求内的下游并发度 + 每个下游独立 timeout，防止某个慢服务拖垮整体；再加 TaskGroup 做失败隔离，允许部分降级。

为什么不用 aiohttp？团队已有 httpx 使用习惯，且 httpx 的 API 和 requests 高度相似，迁移成本低，0.25 版本性能也够用了。aiohttp 性能略好但 API 更啰嗦，这里不纠结。

关键设计点：并发度控制放在哪一层。我的选择是：

• 单请求内：5 个下游天然并发，用 TaskGroup 即可，不需要 semaphore
• 全局：用 Limits(max_connections=200, max_keepalive_connections=50) 控制到每个下游的连接数
• 单下游超时：httpx.Timeout(connect=0.5, read=1.5, write=0.5, pool=0.5)

四、核心实现

4.1 After 代码

# after: app_async.py
import asyncio
import httpx
from fastapi import FastAPI
from contextlib import asynccontextmanager

DOWNSTREAMS = {
    "profile": "http://profile-svc:8001/v1/profile",
    "orders": "http://order-svc:8002/v1/summary",
    "points": "http://points-svc:8003/v1/balance",
    "messages": "http://msg-svc:8004/v1/unread",
    "recommend": "http://rec-svc:8005/v1/list",
}

# 全局单例 client，连接池复用
_client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _client
    _client = httpx.AsyncClient(
        timeout=httpx.Timeout(connect=0.5, read=1.5, write=0.5, pool=0.5),
        limits=httpx.Limits(
            max_connections=200,
            max_keepalive_connections=50,
            keepalive_expiry=30.0,
        ),
        http2=False,  # 下游都是 HTTP/1.1，别开
    )
    yield
    await _client.aclose()

app = FastAPI(lifespan=lifespan)

async def fetch_one(name: str, url: str, user_id: int) -> tuple[str, dict]:
    try:
        resp = await _client.get(url, params={"user_id": user_id})
        resp.raise_for_status()
        return name, resp.json()
    except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
        # 单个下游失败不拖垮整体，返回降级值
        return name, {"_degraded": True, "reason": type(e).__name__}

@app.get("/v1/aggregate/{user_id}")
async def aggregate(user_id: int):
    result: dict = {}
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(fetch_one(name, url, user_id))
            for name, url in DOWNSTREAMS.items()
        ]
    for t in tasks:
        name, data = t.result()
        result[name] = data
    return result

几个细节值得说：

• lifespan 里创建 client：不要在模块顶层 AsyncClient()，那样会绑定到 import 时的 event loop，uvicorn 多 worker 或测试时会炸。用 lifespan 最稳。
• TaskGroup 而非 gather：gather 默认 return_exceptions=False，一个任务抛异常其他任务不会被取消，容易出现"部分成功但整体 500"的尴尬。TaskGroup 在任一子任务异常时会取消其余任务并抛出 ExceptionGroup，语义更清晰。不过我这里在 fetch_one 内部就 catch 了，所以 TaskGroup 主要起结构化并发的作用。
• http2=False：很多人一看 httpx 支持 HTTP/2 就开，但下游全是 HTTP/1.1 服务，开了反而多一层协商开销，实测 QPS 掉了约 8%。

4.2 启动命令

# before
gunicorn -w 8 -b 0.0.0.0:8000 app_sync:app

# after
uvicorn app_async:app --host 0.0.0.0 --port 8000 \
  --workers 4 --loop uvloop --http httptools

--workers 4 对应 4 核，每个 worker 一个 event loop。uvloop 和 httptools 是 uvicorn 的性能标配，实测比默认 asyncio loop + h11 组合 QPS 高约 25%。注意：async 服务不是 worker 越多越好，单 worker 已经能吃满一个核的 IO 并发，worker 数 ≈ CPU 核数即可。

五、踩坑与优化

这部分是本文最有价值的地方，坑都是真金白银踩出来的。

坑 1：连接池打满导致 P99 抖动。 最初 max_connections 设了默认值 100，压测到 QPS 1500 时开始出现 PoolTimeout。原因是 5 个下游共用一个 client，连接被 5 个 host 分摊，实际每个 host 只有 20 条连接可用。改成 200 后消失。经验值：max_connections ≈ 目标QPS × 单请求平均耗时(s) / 下游数，留 1.5 倍余量。

坑 2：requests 的 timeout 语义和 httpx 不一样。 requests 的 timeout=3 是"连接+读取"的总超时上限（单次 socket 操作），httpx 的 Timeout 是分阶段的。迁移时直接照抄 timeout=3 会变成四阶段各 3 秒，最坏情况 12 秒。必须显式拆成 connect/read/write/pool，我最终用的是 0.5/1.5/0.5/0.5。

坑 3：同步代码混进 async 函数。 重构初期有人在 fetch_one 里加了一行 json.loads(...) 还好，但有人加了 requests.get 做兜底，直接把 event loop 阻塞了，QPS 断崖式掉到 200。教训：async 函数里禁止任何同步 IO，上线前用 flake8-async 或 code review checklist 卡住。

坑 4：日志/监控用了同步库。 原来用的日志 handler 是同步写文件，高并发下成了新瓶颈。换成 logging.handlers.QueueHandler + 独立线程消费后解决。监控埋点从 prometheus_client 的同步 Counter 改成批量上报。

优化 1：下游结果缓存。 recommend 和 profile 这两个下游数据变化慢（TTL 60s），加了进程内 cachetools.TTLCache，命中率约 35%，进一步把 P99 压下去一截。注意 asyncio 下单进程缓存要注意内存上限，maxsize 必须设。

优化 2：降级策略。 TaskGroup 的语义是"一个失败全取消"，但这不符合业务预期——推荐挂了不该让整个首页 500。所以我在 fetch_one 内部做了 try/except 降级，返回 {"_degraded": True}，BFF 层识别后走兜底 UI。这才是"隔离"的正确姿势：并发编排负责快，降级逻辑负责稳。

六、效果数据

压测条件：wrk，-t4 -c100 -d60s，同一批 user_id 随机，下游服务 mock 固定延迟（profile 120ms、orders 90ms、points 60ms、messages 80ms、recommend 150ms）。

几点说明：

• QPS 15 倍提升里，大约 8 倍来自并发化（串行 500ms → 并发取最慢的 150ms），约 2 倍来自连接池复用（省掉 TCP/TLS 握手），剩下来自 uvloop 和 Python 3.11 的调度优化。
• P99 从 2.3s 降到 210ms，主要归功于超时隔离——以前一个慢请求会拖住 worker，现在单下游 1.5s 超时直接降级，不传染。
• 内存涨了 140MB，是连接池 + 缓存 + 4 个 event loop 的代价，4 核 8G 机器完全可接受。
• 注意 QPS 1800 是 mock 下游的成绩，接真实下游（有长尾）预计在 1200~1500 之间，我们灰度后线上实测稳定在 1400 左右。

七、总结

这次重构的核心不是"用了 asyncio"，而是识别出这是 IO-bound 场景，然后把等待时间重叠起来。同步模型下你加 8 个 worker，本质是"用 8 份资源模拟 8 个并发"，而 asyncio 单进程就能撑起上千并发连接，资源效率差一个数量级。

给准备做类似改造的同学几条实在建议：

1. 先确认是 IO-bound。如果 CPU 打满，asyncio 帮不了你，该上多进程/多机。
2. 版本别省。Python 3.11+ + httpx 0.25+ + uvloop，这套组合的稳定性比两年前好太多，别在 3.8 上折腾。
3. 连接池和超时是第一优先级，比换框架影响还大。我见过不少人换了 FastAPI 但 client 每次请求新建，QPS 反而没涨。
4. 降级逻辑必须内建。异步并发把失败放大得也快，一个下游抖动可能瞬间打出大量异常，隔离和兜底要在设计阶段就想好。
5. 别迷信 worker 数。ASGI 服务 worker 数量 ≈ CPU 核数，多了是浪费内存。

最后一句：asyncio 不是银弹，它把"等待"这件小事从线程模型里解放出来，代价是你得对 event loop 的阻塞点保持警惕。只要守住"async 函数里不碰同步 IO"这条底线，收益是实打实的。代码已开源到内部仓库 aggregator-api 的 feat/async 分支，欢迎来拍砖。