用asyncio重构同步Web API：QPS从180提升到2100的实践

一、问题背景：一个被IO拖死的查询接口

事情是这样的。我们有个内部聚合查询服务，对外提供一个 /api/v1/aggregate 接口，用户传一个ID进来，服务要同时去三个下游拿数据：用户中心（拿基础信息）、订单服务（拿最近订单）、风控服务（拿风险标签），然后拼装返回。

原来的代码很朴素，同步 requests 顺序调用：

# before.py  同步版本
import requests
from flask import Flask, jsonify

app = Flask(__name__)

TIMEOUT = 5  # 秒

def fetch_user(uid):
    return requests.get(f"http://user-svc/users/{uid}", timeout=TIMEOUT).json()

def fetch_orders(uid):
    return requests.get(f"http://order-svc/orders?uid={uid}", timeout=TIMEOUT).json()

def fetch_risk(uid):
    return requests.get(f"http://risk-svc/risk/{uid}", timeout=TIMEOUT).json()

@app.route("/api/v1/aggregate/")
def aggregate(uid):
    user = fetch_user(uid)
    orders = fetch_orders(uid)
    risk = fetch_risk(uid)
    return jsonify({"user": user, "orders": orders, "risk": risk})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

三个下游的平均响应分别是 80ms、120ms、60ms，加起来 260ms 打底。上线后在4核8G的机器上跑 gunicorn（4 worker），单机 QPS 卡在 180 左右，P99 延迟 1.2s——因为请求一多，worker 全被阻塞在等 IO 上，队列越堆越长。

问题的本质很清楚：这三个调用彼此独立，却用同步方式串行执行，而且每个 worker 在等待网络返回时完全占着线程不干活。Python 的 GIL + 同步 IO 组合，在这种 IO 密集场景下就是灾难。

二、环境与版本

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

• Python 3.11.6（3.11 的 asyncio 在性能上有明显优化，比 3.8 快不少）
• aiohttp 3.9.3（异步 HTTP 客户端）
• aiohttp-jinja2 不需要，这里纯 JSON
• Flask 3.0.2（保留原有框架，用 async 视图 + asgiref 或直接换 ASGI）
• 压测工具：wrk 4.2.0，wrk -t4 -c200 -d60s
• 机器：4核8G，下游服务部署在同机房，网络 RTT < 1ms

这里有个决策点：Flask 本身是 WSGI 同步框架，不能直接跑 async 视图。我评估了两个方案：

1. 保留 Flask，用 asgiref.sync.async_to_sync 把 async 函数包一层——能跑，但每个请求仍然占用一个 WSGI 线程，只是把并发转移到了内部事件循环，收益有限；
2. 迁移到 FastAPI（ASGI）+ uvicorn——改动稍大，但能真正发挥异步优势。

考虑到这个服务接口不多，我选了方案2，迁移成本其实很低。下面代码用 FastAPI 展示，但核心的 asyncio 逻辑在两种框架下是一样的。

三、方案设计

核心思路三条：

1. 把串行改成并发：三个下游调用用 asyncio.gather 并发发起，总耗时从"三者之和"降为"三者最大值"，即 260ms → 120ms。
2. 共享连接池：aiohttp 的 ClientSession 内部维护连接池，复用到下游的 TCP 连接，避免每次握手。设置 connector limit=200，limit_per_host=100。
3. 合理设置超时与并发上限：用 asyncio.wait_for 或 aiohttp 的 ClientTimeout 控制单请求超时，用 asyncio.Semaphore 限制同时打向下游的并发数，防止把下游打挂。

还有一个容易忽略的点：单个请求内的并发不等于服务整体的并发。asyncio 是单线程事件循环，一个请求的协程在等 IO 时会主动让出，其他请求的协程才能跑。所以真正的吞吐提升来自"事件循环在等 IO 期间能处理更多请求"，而不是"一个请求变快了"——虽然两者都成立。

四、核心实现

after 版本代码：

# after.py  asyncio 异步版本
import asyncio
import aiohttp
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

# 全局连接池，应用生命周期内复用
session: aiohttp.ClientSession | None = None

# 限制打向下游的总并发，保护下游
DOWNSTREAM_SEM = asyncio.Semaphore(300)

# 单请求超时配置
TIMEOUT = aiohttp.ClientTimeout(total=3, connect=1, sock_read=2)

async def fetch_json(url: str):
    async with DOWNSTREAM_SEM:
        async with session.get(url, timeout=TIMEOUT) as resp:
            resp.raise_for_status()
            return await resp.json()

async def fetch_user(uid: str):
    return await fetch_json(f"http://user-svc/users/{uid}")

async def fetch_orders(uid: str):
    return await fetch_json(f"http://order-svc/orders?uid={uid}")

async def fetch_risk(uid: str):
    return await fetch_json(f"http://risk-svc/risk/{uid}")

@app.on_event("startup")
async def startup():
    global session
    connector = aiohttp.TCPConnector(
        limit=200,
        limit_per_host=100,
        ttl_dns_cache=300,
        enable_cleanup_closed=True,
    )
    session = aiohttp.ClientSession(connector=connector)

@app.on_event("shutdown")
async def shutdown():
    await session.close()

@app.get("/api/v1/aggregate/{uid}")
async def aggregate(uid: str):
    try:
        user, orders, risk = await asyncio.gather(
            fetch_user(uid),
            fetch_orders(uid),
            fetch_risk(uid),
        )
    except asyncio.TimeoutError:
        return JSONResponse({"error": "downstream timeout"}, status_code=504)
    except aiohttp.ClientError as e:
        return JSONResponse({"error": str(e)}, status_code=502)
    return {"user": user, "orders": orders, "risk": risk}

启动命令：

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

注意 --loop uvloop，uvloop 能把事件循环性能再提一截，实测在这个场景下 QPS 大约多 10%~15%。--workers 4 对应4核，每个 worker 一个独立事件循环，充分利用多核（asyncio 本身只用单核）。

几个关键参数解释：

• limit=200：整个 ClientSession 的连接池上限，超过会排队等待；
• limit_per_host=100：单个下游主机的连接上限，避免对某一个下游建立过多连接；
• ClientTimeout(total=3, connect=1, sock_read=2)：总超时3s，建连1s，读2s，分层控制便于定位问题；
• Semaphore(300)：全局并发闸门，300 是压测调出来的——再高下游开始出现超时，再低吞吐上不去。

五、踩坑与优化

坑1：事件循环里混入同步阻塞调用。 第一次改完发现 QPS 只涨到 400，远低于预期。用 py-spy 抓了一下，发现 JSON 序列化那里用了某个同步库，虽然只有几毫秒，但在事件循环里就是硬阻塞，所有协程都得等。换成 orjson 后恢复正常。教训：异步服务里任何超过 1ms 的同步操作都要警惕，实在避不开就用 loop.run_in_executor 丢到线程池。

坑2：连接池泄漏。 早期版本我在每个请求里 async with aiohttp.ClientSession() 新建 session，压测跑几分钟后报 Too many open files。改成全局单例 session + startup/shutdown 生命周期管理后解决。aiohttp 官方文档也明确建议复用 session。

坑3：gather 没有异常隔离。 asyncio.gather 默认一个协程抛异常，其他协程的结果就丢了（除非 return_exceptions=True）。我们的场景是三个数据缺一不可，所以保持默认行为、直接返回错误是对的；但如果某个下游可以降级，就要用 return_exceptions=True 单独处理。

坑4：超时没设导致雪崩。 最早没设 timeout，某个下游抖动时请求全部挂起，事件循环里堆积大量协程，内存飙升。加上分层 timeout 后，超时的请求快速失败，系统保持稳定。

优化点： 对同一个 uid 的重复请求加了一层进程内 LRU 缓存（cachetools.TTLCache，maxsize=10000，ttl=5s），命中率约 12%，进一步降低了 P99。

六、效果数据

压测条件一致：wrk -t4 -c200 -d60s，同一台4核8G机器，下游服务不变。

QPS 提升约 11.7 倍，P99 从 1.2s 降到 212ms。需要说明的是，这个提升里既有"并发调用缩短单请求耗时"的贡献，也有"事件循环在高并发下不浪费线程"的贡献，后者是大头。另外错误率下降主要是超时配置的功劳，不是异步本身的。

再多说一句：异步不是银弹。如果下游是 CPU 密集或者有同步阻塞依赖，asyncio 反而会放大问题。我们这个场景是典型的 IO 密集、下游独立，才适合这么改。

七、总结

这次改造的核心其实就三点：识别出串行 IO 是瓶颈、用 asyncio.gather 做请求内并发、用全局 ClientSession + 连接池 + 分层超时保证稳定性。代码量不大，但每一步都有坑——尤其是同步阻塞混入事件循环和连接池泄漏这两个，不看监控根本发现不了。

如果你也在维护同步 Web 服务，建议先做一件事：用 py-spy 或者简单的耗时打点，确认瓶颈确实在 IO 等待上，再决定要不要上 asyncio。如果瓶颈在 CPU 或者数据库慢查询，改异步只会让你多一层调试成本。

最后附上我用的依赖清单，方便复现：

fastapi==0.110.0
uvicorn[standard]==0.27.1
aiohttp==3.9.3
uvloop==0.19.0
orjson==3.9.13
cachetools==5.3.3

有问题欢迎评论区交流，尤其是连接池参数调优那块，不同下游特性差别挺大的，我的 300 并发闸门不一定适合你。