asyncio重构Web API：并发从47提升到2100 QPS

一、问题背景：一个"看起来没毛病"的接口

事情是这样的。我们有个商品详情聚合接口 /api/v1/product/{id}，逻辑很朴素：拿商品ID，分别去用户服务查卖家信息、去订单服务查最近成交、去库存服务查实时库存，拼一起返回。

代码大概长这样（简化版）：

# before: app/api/product.py
import requests
from fastapi import FastAPI

app = FastAPI()

USER_SVC = "http://user-svc:8001"
ORDER_SVC = "http://order-svc:8002"
STOCK_SVC = "http://stock-svc:8003"

@app.get("/api/v1/product/{pid}")
def get_product(pid: int):
    user = requests.get(f"{USER_SVC}/user/by-product/{pid}", timeout=2).json()
    orders = requests.get(f"{ORDER_SVC}/recent/{pid}", timeout=2).json()
    stock = requests.get(f"{STOCK_SVC}/stock/{pid}", timeout=2).json()
    return {"pid": pid, "seller": user, "recent_orders": orders, "stock": stock}

本地测没问题，上线后监控也"看起来还好"——平均延迟 120ms 左右。直到大促前压测，P99 直接飙到 380ms，QPS 卡在 47 上不去，一看机器 CPU 才 8%，内存也闲得很。

这就是典型的 I/O 阻塞型瓶颈：三个下游调用是串行的，每个平均 40ms，加起来 120ms 起步，而且每个请求都占着一个 worker 线程干等网络。FastAPI 默认跑在 uvicorn 上，但因为我用的是 def 而不是 async def，它会被丢到 threadpool 里执行，threadpool 默认 40 个线程（anyio 的默认值），40 个线程全在等 I/O，新请求就得排队。

47 QPS ≈ 40 线程 / (120ms 串行 + 排队开销)，算得通。

二、环境与版本

先把家伙什列清楚，后面所有数字都基于这套：

• Python 3.11.6
• FastAPI 0.109.0
• uvicorn 0.27.0（--workers 4，--loop uvloop）
• httpx 0.26.0（异步客户端）
• 下游服务：3 个内部 HTTP 服务，单次 P50 约 35-45ms
• 压测机：wrk 4.2.0，-t 8 -c 200 -d 60s
• 被测机：4C8G 容器，同一内网

改造前用的是 requests 2.31.0。

三、方案设计：async 化 + 并发聚合 + 限流

思路不复杂，但要拆成三层：

第一层：把 def 改成 async def，把 requests 换成 httpx.AsyncClient。 这一步是让请求真正跑在事件循环上，而不是 threadpool。

第二层：三个下游调用没有依赖关系，用 asyncio.gather 并发。 串行 120ms → 并发取 max(40ms) ≈ 45ms。

第三层：加 asyncio.Semaphore 做下游保护。 并发上去了，如果下游某个服务抖动，雪崩会更快。给每个下游服务一个信号量，限制同时打过去的请求数，配合 httpx 的连接池上限。

关于客户端生命周期，我一开始图省事在函数里 async with httpx.AsyncClient()，每次请求新建连接池——这是个大坑，后面踩坑小节细说。正确做法是全局单例 + lifespan 管理。

四、核心实现

先看改造后的完整代码：

# after: app/api/product.py
import asyncio
import httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException

USER_SVC = "http://user-svc:8001"
ORDER_SVC = "http://order-svc:8002"
STOCK_SVC = "http://stock-svc:8003"

# 每个下游一个信号量，限制并发打过去的请求数
SEM_USER = asyncio.Semaphore(200)
SEM_ORDER = asyncio.Semaphore(200)
SEM_STOCK = asyncio.Semaphore(100)  # 库存服务比较脆，压小一点

class Downstream:
    def __init__(self):
        self.client: httpx.AsyncClient | None = None

    async def start(self):
        # 连接池参数很关键，见下文
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(2.0, connect=0.5),
            limits=httpx.Limits(
                max_connections=300,
                max_keepalive_connections=100,
                keepalive_expiry=30.0,
            ),
            headers={"Connection": "keep-alive"},
        )

    async def stop(self):
        if self.client:
            await self.client.aclose()

downstream = Downstream()

@asynccontextmanager
async def lifespan(app: FastAPI):
    await downstream.start()
    yield
    await downstream.stop()

app = FastAPI(lifespan=lifespan)

async def fetch_user(pid: int):
    async with SEM_USER:
        r = await downstream.client.get(f"{USER_SVC}/user/by-product/{pid}")
        r.raise_for_status()
        return r.json()

async def fetch_orders(pid: int):
    async with SEM_ORDER:
        r = await downstream.client.get(f"{ORDER_SVC}/recent/{pid}")
        r.raise_for_status()
        return r.json()

async def fetch_stock(pid: int):
    async with SEM_STOCK:
        r = await downstream.client.get(f"{STOCK_SVC}/stock/{pid}")
        r.raise_for_status()
        return r.json()

@app.get("/api/v1/product/{pid}")
async def get_product(pid: int):
    try:
        # return_exceptions=False：任一失败直接抛，由外层统一处理
        user, orders, stock = await asyncio.gather(
            fetch_user(pid),
            fetch_orders(pid),
            fetch_stock(pid),
        )
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=502, detail=f"downstream error: {e.response.status_code}")
    except (httpx.ConnectTimeout, httpx.ReadTimeout) as e:
        raise HTTPException(status_code=504, detail=f"downstream timeout: {type(e).__name__}")
    return {"pid": pid, "seller": user, "recent_orders": orders, "stock": stock}

启动命令：

uvicorn app.main:app --host 0.0.0.0 --port 8080 \
  --workers 4 --loop uvloop --http httptools --no-access-log

几个参数说明下为什么这么选：

• max_connections=300：单 worker 最多 300 个 TCP 连接，4 个 worker 就是 1200，够 200 并发打满。太小会导致排队等连接，太大容易把下游打挂。
• keepalive_expiry=30.0：下游服务的 nginx keepalive_timeout 是 60s，设 30s 保证连接在被对端关掉前主动回收，避免 Connection reset by peer。
• connect=0.5：内网连接 500ms 超时足够了，快速失败比傻等好。
• SEM_STOCK=100：库存服务是共享的、比较脆，单独压小，这是业务上的保护，不是技术上的。

另外注意 asyncio.gather 默认 return_exceptions=False，第一个异常就抛出来。但有个细节：gather 抛异常时，其它协程并不会被自动取消，它们会继续跑完（只是结果被丢弃）。如果需要"一个失败全部取消"，得用 TaskGroup（Python 3.11+）或者手动 asyncio.wait(FIRST_EXCEPTION)。这里下游都是幂等查询，让它跑完无所谓，所以没处理。

五、踩坑与优化

坑 1：每次请求新建 AsyncClient。 第一版我写的是 async with httpx.AsyncClient() as client:，压测 QPS 反而比改之前还低。原因很简单：每次新建连接池就要重新 TCP 三次握手 + TLS（我们内部虽然是 HTTP，但握手也是成本），连接复用完全没了。改成 lifespan 管理全局单例后，QPS 直接翻倍。这个坑很经典，httpx 官方文档其实有写，但我当时没仔细看。

坑 2：asyncio.Semaphore 在跨事件循环复用。 一开始我把 semaphore 定义成模块级变量，Python 3.10 之前 semaphore 会绑定创建时的 loop，多 worker 或者测试环境里复用会报 got Future attached to a different loop。Python 3.10+ 好一些，但稳妥起见还是在 lifespan 里创建，或者确保模块级创建时机在 loop 起来之后。我们跑 3.11，实测模块级没问题，但为了清晰还是保留了模块级（semaphore 本身不依赖 loop，只有内部 waiter 才依赖）。

坑 3：超时没设，一个慢下游拖垮全局。 改造前 requests 有个 timeout=2，改造后我一开始忘了给 httpx 设超时，默认是 5s 而且只是总超时。结果库存服务某次抖动 4s，gather 就等 4s，P99 直接爆。后来改成 httpx.Timeout(2.0, connect=0.5)，并且把库存的 semaphore 压到 100，双保险。

坑 4：worker 数量和 semaphore 的关系。 我一开始以为 semaphore 是全局的，设 200 就是全机器 200 并发。实际是 每个 worker 进程独立一份，4 个 worker 就是 800 并发。这个认知偏差导致我一开始把 semaphore 设太小，白白限制了吞吐。后来按"单 worker 并发上限"来算，才调对。

优化：uvloop + httptools。 默认 asyncio 的 loop 性能一般，换 uvloop 后同压测条件下 QPS 大概再涨 15% 左右。httptools 是 uvicorn 的 HTTP 解析器，比默认的 h11 快，--http httptools 开上。

六、效果数据

压测条件统一：wrk -t 8 -c 200 -d 60s，4 worker，同一台 4C8G 容器，下游服务不变。

P99 从 380ms 到 62ms，主要收益来自并发聚合（串行 120ms → 并发 45ms）加上连接复用减少了握手开销。QPS 涨 44 倍，是因为原来 40 个 threadpool 线程全在阻塞等 I/O，现在一个事件循环能同时挂几千个协程。

CPU 从 8% 涨到 55%，这个不是"变差了"，是原来根本没把机器用起来。55% 是个比较健康的水位，再往上压 QPS 还能涨，但下游先扛不住了——压到 2500 QPS 时库存服务开始超时，所以 2100 是我们认为的安全水位。

内存涨了 50MB，主要是连接池和协程栈，可以接受。

七、总结

这次改造说复杂不复杂，核心就三件事：async 化、并发聚合、连接池复用。但真正让数字好看的是那些细节——全局 client 单例、超时分层设置、semaphore 按 worker 算、uvloop 加速。

几个可以直接抄走的经验：

1. FastAPI 里凡是调外部 HTTP 的接口，一律 async def + httpx.AsyncClient，别用 def + requests，threadpool 是隐形的并发天花板。
2. AsyncClient 必须是全局单例，用 lifespan 管理生命周期，别在函数里 async with。
3. asyncio.gather 适合无依赖的并发聚合，但要注意异常时其它协程不会自动取消；需要"全有或全无"语义就用 3.11 的 TaskGroup。
4. semaphore 是 per-process 的，算并发上限要乘以 worker 数。
5. 超时一定要分层设：connect 短、read 适中、总超时兜底。没有超时的异步代码比同步代码更危险，因为你以为它在并发，其实它在堆积。

最后说句实话：异步不是银弹。如果瓶颈在 CPU 或者数据库连接数，async 救不了你。但如果是 I/O 等待型，且下游能扛住并发，那这个改造的性价比极高——我们这次改动不到 100 行代码，机器没加一台，QPS 翻了 40 多倍。