一、问题背景：一个慢得让人失眠的接口

去年底接手的项目里有个订单聚合查询接口 /api/order/detail，逻辑本身不复杂：用户传订单号进来，服务端要依次调用下游三个服务——订单中心、物流服务、用户中心，把结果拼起来返回。

问题在于，这三个调用是串行的，而且用的是 requests 同步库。更要命的是，代码里还嵌套了一层循环：一个订单可能包含多个子订单，每个子订单都要查一次物流。典型的一次请求，最坏情况下要发起 1 + N + 1 次 HTTP 调用，N 平均是 3。

上线后监控显示：QPS 峰值 120，P99 延迟 2300ms，机器 CPU 利用率只有 15%，但线程池经常打满。典型的 IO 密集型任务被同步模型拖死了。

我们的目标很明确：不改下游服务，不动业务逻辑，只改调用方式，把 QPS 拉到 1000 以上，P99 压到 300ms 以内。

二、环境与版本

先说清楚技术栈，避免大家照着抄的时候踩版本坑：

  • Python 3.11.6（3.10+ 才有 asyncio.TaskGroup，后面会用到）
  • 原方案：Flask 2.3.2 + requests 2.31.0 + gunicorn 21.2.0（sync worker，8 workers）
  • 新方案：FastAPI 0.109.0 + httpx 0.26.0 + uvicorn 0.27.0
  • 压测工具：wrk 4.2.0，命令 wrk -t4 -c200 -d30s --latency
  • 机器：4核8G，Ubuntu 22.04，下游服务部署在同一内网，RTT 约 5ms

注意一点：FastAPI 只是顺手换的，核心收益来自 asyncio，你用 aiohttp 裸写也能拿到差不多的数据。

三、方案设计：把串行改成并发

先看改造前的逻辑，简化后大概是这样：

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

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

def get_order(oid):
    return SESS.get(f"http://order-svc/order/{oid}", timeout=1).json()

def get_logistics(oid):
    return SESS.get(f"http://logi-svc/logistics/{oid}", timeout=1).json()

def get_user(uid):
    return SESS.get(f"http://user-svc/user/{uid}", timeout=1).json()

@app.route("/api/order/detail/")
def detail(oid):
    order = get_order(oid)
    sub_orders = order["sub_orders"]
    logistics = [get_logistics(s["id"]) for s in sub_orders]  # 串行
    user = get_user(order["uid"])
    return jsonify({"order": order, "logistics": logistics, "user": user})

4 次下游调用，串行执行，假设每次 20ms，光网络往返就 80ms 起步。并发一上来，gunicorn 的 8 个 sync worker 根本不够用，请求全堵在队列里。

改造思路分三步：

  1. 把同步 HTTP 换成异步 HTTP：requests → httpx.AsyncClient，复用连接池。
  2. 把串行调用改成并发：get_order 必须先返回才能拿到 sub_orders 和 uid，这一步没法省；但之后的物流查询和用户查询可以并发，多个子订单的物流查询也可以并发。
  3. 用 TaskGroup 管理并发任务：Python 3.11 的 asyncio.TaskGroup 比 gather 更好用，异常处理更清晰，任务泄漏也能自动取消。

四、核心实现

改造后的代码：

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

app = FastAPI()

# 全局复用连接池，limits 参数很关键，后面踩坑小节会讲
client = httpx.AsyncClient(
    timeout=httpx.Timeout(1.0, connect=0.3),
    limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
)

async def get_order(oid):
    r = await client.get(f"http://order-svc/order/{oid}")
    return r.json()

async def get_logistics(oid):
    r = await client.get(f"http://logi-svc/logistics/{oid}")
    return r.json()

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

@app.get("/api/order/detail/{oid}")
async def detail(oid: str):
    order = await get_order(oid)  # 必须先拿到 order

    async with asyncio.TaskGroup() as tg:
        user_task = tg.create_task(get_user(order["uid"]))
        logi_tasks = [
            tg.create_task(get_logistics(s["id"]))
            for s in order["sub_orders"]
        ]

    logistics = [t.result() for t in logi_tasks]
    user = user_task.result()
    return {"order": order, "logistics": logistics, "user": user}

关键点解释：

  • TaskGroup 里的任务全部并发执行，任何一个抛异常，其余任务会被自动取消，不用手写 gather(return_exceptions=True) 再过滤。
  • httpx.AsyncClient 全局只创建一次，连接池跨请求复用。如果每个请求都 new 一个 client，性能会比同步还差。
  • Timeout(1.0, connect=0.3) 里 connect 单独设短一点，避免下游挂掉时连接阶段就卡满。

启动命令也换了：

uvicorn app_async:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop

4 个 worker 对应 4 核，uvloop 能把事件循环性能再拉高一截。

五、踩坑与优化

坑1：连接池默认值太小

httpx 默认 max_connections=100，max_keepalive_connections=20。压测到 150 并发时开始出现 PoolTimeout，日志里一堆连接等待超时。把 max_connections 提到 200，max_keepalive 提到 50，问题消失。

坑2：忘了给下游加超时

httpx.Timeout 不设的话默认是 5 秒，下游一个慢请求就能拖垮整个事件循环的吞吐。我们统一设成 1 秒，并且用 connect=0.3 单独限制建连。

坑3：TaskGroup 异常传播

一开始我用 asyncio.gather，某个物流查询 404 时抛异常，其他任务不会取消，白白浪费资源。换成 TaskGroup 后，一个失败全部取消，配合 FastAPI 的异常处理，直接返回 502，语义更干净。

优化：给下游结果加本地缓存

物流信息 5 秒内基本不变，我们加了一层 cachetools.TTLCache，命中率约 40%，QPS 又涨了大概 200。

六、效果数据

同样的机器、同样的下游、同样的压测命令 wrk -t4 -c200 -d30s --latency：

指标 before (Flask+requests) after (FastAPI+httpx) 提升
QPS 120 1800 15x
P50 延迟 620ms 42ms 14.8x
P99 延迟 2300ms 180ms 12.8x
CPU 利用率 15% 68% -
内存 480MB 320MB -33%

有意思的是 CPU 利用率反而上去了，因为之前大部分时间都耗在等 IO 上。这才是 IO 密集型服务该有的样子。

七、总结

这次改造的核心其实就三句话：把同步 IO 换成异步 IO，把串行调用改成并发调用，把连接池参数调对。代码量没增加多少，收益却非常明显。

几点经验：

  • Python 3.11 的 TaskGroup 比 gather 好用，新项目直接用。
  • httpx.AsyncClient 一定全局复用，别在请求里 new。
  • 连接池的 max_connections 要按并发量估算，宁大勿小。
  • 别忘了超时，异步代码里一个慢请求的破坏力比同步更大。

如果你的服务也是 IO 密集型、QPS 卡在几百上不去，不妨先看看是不是同步调用拖了后腿。asyncio 不是银弹，但在这种场景下，它确实能把机器榨干。