asyncio重构同步Web API，QPS从120飙到2100

一、问题背景：一个被同步IO拖死的API

事情是这样的。我们有个内部服务叫 user-profile-aggregator，职责很简单：接收一个用户ID，然后并发调用三个下游服务（用户基础信息、订单统计、积分余额），聚合后返回。听起来是个典型的IO密集型场景对吧？

但它是用 Flask 写的，代码大概长这样：

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

app = Flask(__name__)

@app.route("/profile/")
def get_profile(user_id):
    user = requests.get(f"http://user-svc/users/{user_id}", timeout=2).json()
    orders = requests.get(f"http://order-svc/orders?uid={user_id}", timeout=2).json()
    points = requests.get(f"http://point-svc/points/{user_id}", timeout=2).json()
    return jsonify({"user": user, "orders": orders, "points": points})

部署方式是 Gunicorn + 4个 sync worker，跑在 4C8G 的机器上。问题很明显：三个下游调用是串行的，假设每个下游平均响应 80ms，那这个接口光下游等待就要 240ms。更致命的是，每个 worker 同一时刻只能处理一个请求，4个 worker 意味着并发上限就是 4。

我拉了一下生产监控：

• 单节点 QPS：120（高峰期）
• P99 延迟：1.8s
• CPU 使用率：15%（基本都在等IO）
• 平均下游耗时：user-svc 65ms / order-svc 110ms / point-svc 90ms

CPU 才用了 15%，说明瓶颈根本不在计算，纯粹是被同步阻塞卡死了。这种场景不上 asyncio 简直是浪费机器。

二、环境与版本

改造前先把版本钉死，避免踩兼容性的坑：

• Python：3.11.6（3.10+ 对 asyncio 有持续优化，3.11 的 TaskGroup 很好用）
• FastAPI：0.109.0
• Uvicorn：0.27.0（用 uvicorn[standard]，带 uvloop）
• httpx：0.26.0（异步 HTTP 客户端）
• Gunicorn：21.2.0（作为进程管理器）
• 压测工具：wrk 4.2.0，wrk -t8 -c200 -d30s

需要特别说明：uvloop 一定要装。在 Linux 上它能给 asyncio 带来 20%-30% 的吞吐提升，官方 benchmark 里甚至更高。Uvicorn 的 [standard] extra 会自动带上它。

三、方案设计

整体思路分三层：

1. Web 框架层：Flask → FastAPI，原生支持 async def 路由，底层是 ASGI。
2. HTTP 客户端层：requests → httpx.AsyncClient，配合全局连接池复用 TCP 连接。
3. 并发编排层：三个下游调用从串行改成 asyncio.gather 并发，总耗时从 sum 变成 max。

进程模型上，我用 Gunicorn 起多个 Uvicorn worker（-k uvicorn.workers.UvicornWorker），每个 worker 是单线程事件循环，worker 数量设为 2 * CPU核数 + 1 略少一点，实际用 6 个。因为异步模型下单 worker 就能扛很多并发，不需要开太多。

一个关键设计决策：不要把 AsyncClient 每次请求都新建。每次 async with httpx.AsyncClient() 都会重建连接池，TCP 握手 + TLS 开销全白费。正确做法是全局单例，在应用启动时创建，关闭时释放。

四、核心实现

先看改造后的代码：

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

# 全局客户端，生命周期跟随应用
client: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global client
    # 连接池调优：单host最大100连接，最多保留20个keep-alive
    limits = httpx.Limits(
        max_connections=200,
        max_keepalive_connections=50,
        keepalive_expiry=30.0,
    )
    timeout = httpx.Timeout(2.0, connect=0.5)
    client = httpx.AsyncClient(
        limits=limits,
        timeout=timeout,
        http2=False,  # 下游都是内网HTTP，HTTP/2收益不大
    )
    yield
    await client.aclose()

app = FastAPI(lifespan=lifespan)

async def fetch_user(uid: int):
    r = await client.get(f"http://user-svc/users/{uid}")
    r.raise_for_status()
    return r.json()

async def fetch_orders(uid: int):
    r = await client.get(f"http://order-svc/orders", params={"uid": uid})
    r.raise_for_status()
    return r.json()

async def fetch_points(uid: int):
    r = await client.get(f"http://point-svc/points/{uid}")
    r.raise_for_status()
    return r.json()

@app.get("/profile/{user_id}")
async def get_profile(user_id: int):
    # 三个请求并发发出，总耗时 = max(下游耗时) 而非 sum
    user, orders, points = await asyncio.gather(
        fetch_user(user_id),
        fetch_orders(user_id),
        fetch_points(user_id),
    )
    return {"user": user, "orders": orders, "points": points}

启动命令：

gunicorn main:app \
  -k uvicorn.workers.UvicornWorker \
  -w 6 \
  --bind 0.0.0.0:8000 \
  --backlog 2048 \
  --max-requests 10000 \
  --max-requests-jitter 1000 \
  --timeout 30

几个参数解释一下：

• -w 6：4核机器上开6个 worker，每个都是独立事件循环。
• --backlog 2048：高并发下 accept 队列要够大，默认2048也行，但显式写出来更保险。
• --max-requests 10000：周期性重启 worker，防止内存泄漏累积（虽然我们代码没什么泄漏点，但防御性配置）。
• --timeout 30：worker 被 kill 的阈值，别设太短，异步 worker 卡住通常是死锁。

五、踩坑与优化

这一节是重点，因为改造过程并不是一遍就过的。

坑1：asyncio.gather 默认不取消兄弟任务

asyncio.gather 在某个任务抛异常时，默认不会取消其他任务。这意味着如果 user-svc 挂了抛异常，其他两个请求还在跑，会浪费资源。生产环境应该用 return_exceptions=False（默认）配合 TaskGroup，或者显式处理：

# Python 3.11+ 推荐写法
async def get_profile_v2(user_id: int):
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch_user(user_id))
            t2 = tg.create_task(fetch_orders(user_id))
            t3 = tg.create_task(fetch_points(user_id))
        return {"user": t1.result(), "orders": t2.result(), "points": t3.result()}
    except* httpx.HTTPError as eg:
        # 任一失败，其他自动取消
        raise HTTPException(502, f"downstream failed: {eg.exceptions}")

TaskGroup 是 3.11 引入的，比 gather 更安全，推荐新项目直接上。

坑2：同步库混进异步路由，事件循环直接被阻塞

改造初期我偷懒，在某个 helper 里用了 requests 库，结果发现 QPS 上不去。原因是：requests.get() 是同步阻塞的，在 async def 路由里调用它会阻塞整个事件循环，导致该 worker 上所有并发请求都卡住。

排查方法：用 asyncio 的 debug 模式（PYTHONASYNCIODEBUG=1）启动，超过100ms的同步调用会打警告日志。后来全部换成 httpx，问题消失。

铁律：async 路由里不要出现任何同步阻塞 IO。

坑3：连接池太小导致排队

httpx 默认 max_connections=100，但这是全局的，不是 per-host。我一开始没设，结果下游三台服务共享100连接，压测到 500 并发时开始出现连接等待。调成 max_connections=200, max_keepalive_connections=50 后，P99 从 380ms 降到 230ms。

如果你下游是多个 host，建议用 httpx.Limits 显式控制，别用默认。

坑4：超时设置不合理

httpx 的 Timeout(2.0) 是总超时，但连接超时应该单独设更短，因为连接建立失败通常意味着下游不可达，没必要等 2 秒。我改成 Timeout(2.0, connect=0.5)，下游雪崩时能快速失败，避免请求堆积。

优化：加一层内存缓存

user-svc 的用户基础信息变化频率很低，我在聚合层加了个 5 秒的 TTL 缓存（用 cachetools.TTLCache），命中率大概 35%，进一步降低了 P99。

六、效果数据

压测环境：4C8G 云主机，下游用 mock 服务模拟，响应时间固定 user 65ms / order 110ms / point 90ms。

wrk 命令：wrk -t8 -c200 -d30s --latency http://localhost:8000/profile/123

CPU 从 15% 涨到 68% 是好事情——说明机器终于被用起来了。QPS 提升 17.5x 主要来自两个因素：并发调用下游（240ms → 110ms 的均值）+ 单 worker 高并发（4 → 数百）。

生产环境上线后，用 2 台机器替换了原来的 5 台，月成本降了约 60%，P99 从 1.8s 降到 230ms，用户投诉基本消失。

七、总结

这次改造给我最大的感受是：IO 密集型服务的性能问题，90% 都不是代码写得烂，而是模型选错了。同步阻塞模型下，你优化得再极致，一个 worker 也只能同时处理一个请求，天花板就在那里。

asyncio 不是银弹，它适合的场景很明确：高并发、IO 密集、下游调用多。如果你的服务是 CPU 密集型（比如图像处理、加密计算），上 asyncio 反而会拖慢——那种场景应该用多进程或者 C 扩展。

最后几条经验送给要踩坑的同学：

1. 用 Python 3.11+，TaskGroup 比 gather 更安全。
2. httpx 的 AsyncClient 一定要全局单例，连接池参数要显式调。
3. 装 uvloop，性能提升立竿见影。
4. async 路由里绝对不能出现同步阻塞调用，debug 模式能帮你抓出来。
5. 上线前用 wrk 或 locust 压一遍，数据比感觉靠谱。

代码已经开源到内部仓库了，有需要的同学可以找我要。下次准备写写怎么用 asyncio 做限流和熔断，感兴趣的话点个关注。