一、问题背景

事情是这样的。我们组维护一个内部叫 agg-api 的服务，职责很单一：接收一个资源ID列表，并发调用下游3~5个微服务（用户画像、订单统计、内容标签），把结果聚合成一个大JSON返回给前端。典型请求要串3个下游，每个下游平均耗时80~150ms。

原来的实现是Flask + requests，代码大概是这个画风：

# before: app.py (Flask + requests)
import requests
from flask import Flask, request, jsonify

app = Flask(__name__)
S = requests.Session()

@app.route("/aggregate", methods=["POST"])
def aggregate():
    ids = request.json["ids"]
    result = []
    for rid in ids:                      # 串行遍历
        u = S.get(f"http://user-svc/profile/{rid}", timeout=2).json()
        o = S.get(f"http://order-svc/stat/{rid}", timeout=2).json()
        t = S.get(f"http://tag-svc/tags/{rid}", timeout=2).json()
        result.append({"id": rid, "user": u, "order": o, "tags": t})
    return jsonify(result)

问题一目了然：外层按ID串行，内层3个下游也串行。一个请求带10个ID，理论耗时就是 10 × 3 × 100ms = 3秒。线上P99 1.8秒已经算是下游偶尔快的结果了。

高峰期（每天10点、20点两个尖峰）单实例QPS只有120左右，CPU利用率不到15%——全在等IO。我们靠8台4C8G的机器硬扛，机器成本不说，P99延迟让前端同学天天来找我。

这就是典型的 IO-bound 场景，asyncio 的主场。

二、环境与版本

先交代环境，避免版本差异导致行为不一致：

  • Python 3.11.6（3.11的asyncio在异常处理和TaskGroup上有明显改进，不建议低于3.9）
  • FastAPI 0.109.2 + uvicorn 0.27.0（uvicorn[standard]，带uvloop）
  • httpx 0.26.0（异步HTTP客户端，支持连接池）
  • uvloop 0.19.0（Linux下替换默认event loop，实测有10%~20%提升）
  • 部署：4C8G容器，Gunicorn不需要，直接用 uvicorn --workers 4（4核起4个worker，每worker一个event loop）
  • 压测工具：wrk 4.2.0 + 自写Lua脚本，固定请求体10个ID

下游服务都是内网HTTP，RTT约1ms，单次处理80~150ms，本身也做了限流（每服务2000 QPS），这点后面踩坑会提到。

三、方案设计

改造思路分三层：

1. 请求内并发化。 原来"ID串行 × 下游串行"改成"ID之间并发 + 每个ID的3个下游并发"，用 asyncio.gather 一把梭。理论上单请求耗时从 10×3×100ms 压到约 100ms（最慢下游决定）。

2. 用httpx.AsyncClient做全局连接池。 不要每次请求新建client，那是灾难。按worker维度建一个单例，limits 设 max_connections=200, max_keepalive_connections=50，timeout 分连接/读两段设。

3. 并发度要可控。 无脑gather在ID多的时候会把下游打挂（我们真打挂过一次tag-svc）。用 asyncio.Semaphore 限制单请求内的并发下游调用数，比如32。

伪代码结构：

POST /aggregate
  └─ gather( for each id: fetch_one(id) )
       fetch_one: gather(user, order, tag)  ← 受 Semaphore(32) 保护

四、核心实现

先看改造后的核心代码，这是能直接跑的版本（依赖 fastapi httpx uvicorn）：

# after: main.py (FastAPI + httpx.AsyncClient)
import asyncio
import httpx
from fastapi import FastAPI, Request
from contextlib import asynccontextmanager

MAX_CONCURRENCY = 32          # 单请求内下游并发上限
POOL_LIMITS = httpx.Limits(max_connections=200, max_keepalive_connections=50)
TIMEOUT = httpx.Timeout(connect=0.5, read=2.0, write=1.0, pool=1.0)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 全局单例 client，随 worker 生命周期存活
    app.state.client = httpx.AsyncClient(limits=POOL_LIMITS, timeout=TIMEOUT)
    yield
    await app.state.client.aclose()

app = FastAPI(lifespan=lifespan)

async def fetch_one(client: httpx.AsyncClient, rid: str, sem: asyncio.Semaphore):
    async with sem:  # 控制单请求并发度，保护下游
        u, o, t = await asyncio.gather(
            client.get(f"http://user-svc/profile/{rid}"),
            client.get(f"http://order-svc/stat/{rid}"),
            client.get(f"http://tag-svc/tags/{rid}"),
            return_exceptions=True,   # 单个下游挂了不拖垮整请求
        )
    def safe(r):
        return r.json() if isinstance(r, httpx.Response) and r.status_code == 200 else None
    return {"id": rid, "user": safe(u), "order": safe(o), "tags": safe(t)}

@app.post("/aggregate")
async def aggregate(req: Request):
    body = await req.json()
    ids = body["ids"]
    client = req.app.state.client
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    results = await asyncio.gather(*(fetch_one(client, rid, sem) for rid in ids))
    return {"data": results}

几个关键点解释一下：

  • lifespan 里创建 AsyncClient，保证每个 uvicorn worker 一个 client、一套连接池。千万别在路由函数里 async with httpx.AsyncClient()，那是每次请求重建池，keep-alive全废。
  • return_exceptions=True 是生产必需。改造前有个下游超时会把整个请求500，现在降级成该字段为 null，前端能正常渲染。
  • Semaphore 放在请求级别创建（而不是全局），因为限制的是"单个请求打下游的并发"，不同请求之间互不影响。全局并发由连接池的 max_connections 兜底。

再给一个"想更细粒度控制"的版本，用 TaskGroup（3.11+），好处是异常传播语义更清晰，配合 asyncio.timeout 做整体超时：

# 可选：Python 3.11+ TaskGroup + 整体超时
async def fetch_one_v2(client, rid, sem):
    async with sem:
        async with asyncio.timeout(2.5):        # 整条链路硬超时
            async with asyncio.TaskGroup() as tg:
                tu = tg.create_task(client.get(f"http://user-svc/profile/{rid}"))
                to = tg.create_task(client.get(f"http://order-svc/stat/{rid}"))
                tt = tg.create_task(client.get(f"http://tag-svc/tags/{rid}"))
        return {"id": rid, "user": tu.result().json(),
                "order": to.result().json(), "tags": tt.result().json()}

注意 TaskGroup 里任一任务抛异常会取消其余任务并抛 ExceptionGroup，所以这个版本适合"下游必须全成功"的场景；我们聚合接口允许部分失败，线上用的还是 gather(return_exceptions=True) 那版。

五、踩坑与优化

这一段是重点，光把代码改成async，QPS不会自动起飞。

坑1：在async路由里调同步库，直接把event loop堵死。 第一版我图省事，把原来的 requests 调用原封不动塞进 async def，结果QPS不升反降，从120掉到90。原因很简单：requests.get 是阻塞的，在协程里执行会卡住整个event loop，所有并发请求排队。后来全换 httpx.AsyncClient。同理，任何 time.sleep、同步DB驱动（比如 pymysql）、requests 都不能出现在async函数里，要么换异步库，要么 await asyncio.to_thread(...) 丢线程池。

坑2：连接池太小，日志里全是 PoolTimeout。 一开始 max_connections 设了20，压测到600 QPS时开始刷 httpx.PoolTimeout。算一下：单请求10个ID × 3个下游 = 30个并发连接需求，池子20根本不够。调到200后消失。经验公式：max_connections ≥ workers × 单请求峰值并发数，宁大勿小，反正空闲连接会因 keepalive_expiry（默认5s）回收。

坑3：把tag-svc打挂了。 上线第二天，tag-svc的负责人来找我，说他们QPS从800飙到4000。原因是并发化之后，我们的瞬时并发放大了十几倍。加了 Semaphore(32) 后，再配合下游的限流配置对齐，才稳住。教训：异步化会放大下游压力，必须同步和下游对齐容量，这不是你一个服务的事。

坑4：uvloop没开。 一开始直接用 uvicorn main:app --workers 4，后来加 --loop uvloop（uvicorn[standard] 自带），同样的压测脚本，QPS从950涨到1100，P99从220ms降到180ms。Linux下白捡的性能，别忘了。

坑5：--workers 不是越大越好。 试过 --workers 8（4核机器），QPS反而降到980，上下文切换和连接池翻倍导致内存涨到6G。最后定在 workers = CPU核数，即4。

最终配置：uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop --http httptools。

六、效果数据

压测条件：wrk，4线程，200连接，请求体固定10个ID，持续5分钟，取稳定段。下游服务mock了固定100ms延迟（用 asyncio.sleep(0.1) 的假服务，保证对比公平）。

指标 Before (Flask+requests, 8台) After (FastAPI+httpx, 2台)
单实例 QPS 120 1100
P50 延迟 820ms 95ms
P99 延迟 1800ms 180ms
单实例 CPU 利用率 13% 68%
单实例内存 380MB 520MB
支撑同等流量所需机器 8 台 4C8G 2 台 4C8G

几个补充观察：

  • 延迟从"ID数×下游数×100ms"的串行模型，变成了接近"最慢下游100ms + 调度开销"的并发模型，ID数从10加到50，P99只从180ms涨到310ms（受Semaphore排队影响），而老版本会线性涨到9秒。
  • 内存涨了140MB，主要是连接池和协程栈，完全可接受。
  • 机器从8台减到2台，按我们内部计价一年省下的钱够组里吃好几顿火锅。

另外提醒一句：以上数字强依赖下游mock延迟稳定。真实环境下游有抖动，return_exceptions=True 的降级率会上升，建议给聚合结果加个"部分成功"标记，让前端能区分。

七、总结

这次改造的核心不是"用了asyncio"，而是三件事：把串行IO改成并发、把连接池做成进程级单例、把并发度显式管起来。asyncio只是提供了表达并发的最低成本语法，真正决定性能的是连接复用和背压控制。

几条可以直接抄走的结论：

  1. IO-bound的聚合类接口，asyncio收益巨大，我们这个场景是9倍QPS。
  2. httpx.AsyncClient 必须全局单例 + 合理 Limits，否则性能还不如同步。
  3. Semaphore 保护下游，异步化前先和下游对齐容量，别做那个把别人打挂的人。
  4. Linux下开uvloop，免费的性能。
  5. workers = CPU核数，别贪多。
  6. 生产代码里 gather(return_exceptions=True) 基本是标配，单点故障不该拖垮整个响应。

如果你的服务也是"一个请求扇出多个下游"的形态，且CPU利用率长期低于30%，那基本可以确定是IO在等你，值得动手。至于纯CPU密集的场景，asyncio帮不上忙，老老实实上多进程或者换语言。