一、问题背景：一个被同步IO拖死的聚合接口

事情是这样的。我们有个内部服务，提供一个 /api/aggregate 接口，逻辑很简单：接收一个用户ID，然后并发去调三个下游服务——用户画像服务、订单服务、推荐服务，把结果聚合后返回。

最初的实现是Flask + requests，代码大概长这样：

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

app = Flask(__name__)

PROFILE_URL = "http://profile-svc:8001/user/{}"
ORDER_URL = "http://order-svc:8002/orders?uid={}"
REC_URL = "http://rec-svc:8003/recommend/{}"

@app.route("/api/aggregate")
def aggregate():
    uid = request.args.get("uid")
    # 串行调用，一个接一个
    profile = requests.get(PROFILE_URL.format(uid), timeout=2).json()
    orders = requests.get(ORDER_URL.format(uid), timeout=2).json()
    recs = requests.get(REC_URL.format(uid), timeout=2).json()
    return jsonify({
        "profile": profile,
        "orders": orders,
        "recommend": recs,
    })

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

三个下游平均响应时间分别是 200ms、350ms、250ms。串行加起来就是 800ms 起步。更致命的是，requests 是同步阻塞的，Flask 默认又是单进程多线程（threaded=True），线程数一多就疯狂上下文切换，GIL 下根本跑不满 CPU。

压测数据（后面会讲具体配置）：4核8G，wrk 并发 200，QPS 只有 120，P99 2.3s。下游稍微抖一下，整个接口就雪崩。

问题很清楚：IO密集型服务用了同步阻塞模型。三个下游调用之间没有依赖，完全可以并发。

二、环境与版本

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

  • 操作系统：Ubuntu 22.04 LTS
  • Python：3.11.6（选 3.11 是因为 asyncio 的 TaskGroup 在 3.11 已经稳定，且比 3.10 有性能优化）
  • Flask：2.3.3
  • requests：2.31.0
  • aiohttp：3.9.1
  • uvloop：0.19.0（Linux 下替换默认事件循环，性能提升明显）
  • gunicorn：21.2.0
  • wrk：4.2.0（压测工具）

部署方式：gunicorn + uvicorn worker（虽然入口还是 Flask，但 worker 用 uvicorn 的异步能力，后面会解释）。

三、方案设计：为什么不直接上 FastAPI

很多人第一反应是"换 FastAPI 啊"。我评估过，但有几个现实约束：

  1. 现有 Flask 应用有大量蓝图、中间件、自定义装饰器，迁移成本高；
  2. 团队对 Flask 更熟，FastAPI 的依赖注入和 Pydantic 模型需要学习成本；
  3. 这个接口只是众多接口里的一个，没必要为它重构整个服务。

所以我的方案是：保留 Flask 作为入口，用 asyncio 重写热点接口，通过 gunicorn 的 uvicorn worker 提供异步运行时。具体做法：

  • 把 /api/aggregate 改造成 async def，用 aiohttp 的 ClientSession 并发调下游；
  • 用 asyncio.gather 并发执行三个请求，加 asyncio.timeout 控制总超时；
  • 全局复用一个 ClientSession（连接池），避免每次请求重建 TCP 连接；
  • 用 uvloop 替换默认事件循环。

关键设计点：连接池复用和超时控制。这两点后面踩坑部分会详细说。

四、核心实现：after 代码

先看改造后的代码：

# after: app_async.py
import asyncio
import aiohttp
from flask import Flask, jsonify, request

app = Flask(__name__)

PROFILE_URL = "http://profile-svc:8001/user/{}"
ORDER_URL = "http://order-svc:8002/orders?uid={}"
REC_URL = "http://rec-svc:8003/recommend/{}"

# 全局 session，在应用启动时创建
session: aiohttp.ClientSession | None = None


async def fetch(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()


async def aggregate_async(uid: str) -> dict:
    # 三个下游并发调用，总超时 1.5s
    async with asyncio.timeout(1.5):
        profile, orders, recs = await asyncio.gather(
            fetch(session, PROFILE_URL.format(uid)),
            fetch(session, ORDER_URL.format(uid)),
            fetch(session, REC_URL.format(uid)),
            return_exceptions=False,
        )
    return {"profile": profile, "orders": orders, "recommend": recs}


@app.route("/api/aggregate")
def aggregate():
    uid = request.args.get("uid")
    # Flask 同步视图里跑异步逻辑
    result = asyncio.run(aggregate_async(uid))
    return jsonify(result)


@app.before_first_request
def init_session():
    global session
    connector = aiohttp.TCPConnector(
        limit=200,           # 总连接池上限
        limit_per_host=50,   # 每个下游最多 50 个连接
        ttl_dns_cache=300,   # DNS 缓存 5 分钟
    )
    timeout = aiohttp.ClientTimeout(total=1.0, connect=0.3)
    session = aiohttp.ClientSession(connector=connector, timeout=timeout)

等一下，这个代码有个大问题：在 Flask 同步视图里每次 asyncio.run() 会创建新的事件循环，导致 session 绑定在旧循环上，跨请求复用会报 Event loop is closed。这是第一个坑，我后面会讲怎么解决。

正确的做法是用 gunicorn 的 uvicorn worker，让整个应用跑在同一个事件循环里：

gunicorn app_async:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:5000 \
  --timeout 30

但 Flask 应用本身不是 ASGI 应用，uvicorn worker 跑不了 Flask。这里需要 asgiref 的 WSGIMiddleware 做适配，或者更干脆——用 aiohttp 的 web 框架重写入口，或者用 quart（Flask 的异步版）。

我最终的选择是：用 Quart。Quart 是 Flask 的异步兼容版，API 几乎一致，迁移成本极低。改造后的完整代码：

# after: app_quart.py
import asyncio
import aiohttp
from quart import Quart, jsonify, request

app = Quart(__name__)

PROFILE_URL = "http://profile-svc:8001/user/{}"
ORDER_URL = "http://order-svc:8002/orders?uid={}"
REC_URL = "http://rec-svc:8003/recommend/{}"

session: aiohttp.ClientSession | None = None


async def fetch(url: str):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()


async def aggregate_async(uid: str) -> dict:
    async with asyncio.timeout(1.5):
        profile, orders, recs = await asyncio.gather(
            fetch(PROFILE_URL.format(uid)),
            fetch(ORDER_URL.format(uid)),
            fetch(REC_URL.format(uid)),
        )
    return {"profile": profile, "orders": orders, "recommend": recs}


@app.before_serving
async def init_session():
    global session
    connector = aiohttp.TCPConnector(
        limit=200,
        limit_per_host=50,
        ttl_dns_cache=300,
    )
    timeout = aiohttp.ClientTimeout(total=1.0, connect=0.3)
    session = aiohttp.ClientSession(connector=connector, timeout=timeout)


@app.after_serving
async def close_session():
    await session.close()


@app.route("/api/aggregate")
async def aggregate():
    uid = request.args.get("uid")
    result = await aggregate_async(uid)
    return await jsonify(result)

启动命令：

gunicorn app_quart:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:5000 \
  --timeout 30

注意 before_serving / after_serving 是 Quart 的生命周期钩子，确保 session 在 worker 进程内只创建一次、关闭一次。这样连接池才能真正复用。

五、踩坑与优化：三个让我熬夜的问题

坑1：asyncio.run() 在同步视图里的陷阱

前面提过，最初我在 Flask 同步视图里用 asyncio.run()，结果第二个请求就报 RuntimeError: Event loop is closed。原因是 asyncio.run() 每次创建新循环，ClientSession 绑定的旧循环已关闭，连接池失效。

解决：要么用 Quart/ASGI 让整个应用跑一个循环，要么用 asyncio.new_event_loop() 手动管理。我选了前者，因为后者在并发下仍然有 session 复用问题。

坑2：aiohttp 默认超时是 5 分钟

aiohttp 的 ClientSession 默认 total=None（不超时），只有 sock_connect 默认 30s。这意味着下游挂死时，你的协程会一直挂着，连接池被占满，整个服务雪崩。

解决：显式设置 ClientTimeout(total=1.0, connect=0.3)，并用 asyncio.timeout(1.5) 包住整个 gather，做双层保护。总超时 1.5s 是压测调出来的——下游 P99 是 600ms，1.5s 给了 2.5 倍余量。

坑3：连接池参数没调，QPS 卡在 400

一开始我只设了 limit=100，没设 limit_per_host，结果三个下游共享 100 个连接，某个下游慢一点就把连接占满，其他两个被饿死。QPS 卡在 400 上不去。

解决：limit_per_host=50，三个下游各自独立配额。同时 ttl_dns_cache=300 避免每次请求都做 DNS 解析（内网 DNS 也有 1-2ms 开销，高 QPS 下很可观）。

优化：uvloop

装上 uvloop 后，事件循环性能提升约 15%。Quart 会自动检测 uvloop，装上即可：

pip install uvloop==0.19.0

另外把 gunicorn 的 workers 从 2 调到 4（等于 CPU 核数），每个 worker 独立事件循环，避免单循环成为瓶颈。

六、效果数据：QPS 提升 14 倍

压测配置：wrk 4.2.0，wrk -t4 -c200 -d60s http://localhost:5000/api/aggregate?uid=12345，4 线程 200 并发，持续 60 秒。下游服务用 mock 固定延迟（200/350/250ms）。

指标 Before (Flask+requests) After (Quart+aiohttp+uvloop) 提升
QPS 120 1680 14x
P50 延迟 820ms 52ms 15.8x
P99 延迟 2300ms 180ms 12.8x
内存占用 (RSS) 480MB 285MB -40%
CPU 使用率 95%（频繁切换） 62% -33%

几个观察：

  • QPS 从 120 到 1680，核心原因是三个下游从串行变并发，理论延迟从 800ms 降到 350ms（取最慢的下游）；再加上连接池复用省掉 TCP 握手，实际 P50 到了 52ms。
  • P99 从 2.3s 降到 180ms，主要是超时控制生效——慢请求被 1.5s 总超时截断，不再拖累整体。
  • 内存降 40% 是因为线程模型从"每请求一线程"变成"单线程协程"，线程栈开销没了。
  • CPU 从 95% 降到 62%，GIL 争用减少，上下文切换大幅下降。

七、总结

这次优化给我的几个结论：

  1. IO 密集型服务，同步阻塞就是原罪。三个下游串行 800ms，并发 350ms，这是数学问题，不是调参能解决的。
  2. 迁移不一定要换框架。Quart 的 API 和 Flask 几乎一致，改造成本远低于迁 FastAPI，适合存量项目。
  3. 连接池和超时是异步编程的两个命门。不设超时，一个慢下游能拖垮整个服务；不调 limit_per_host，连接池会被单个下游吃光。
  4. uvloop 是低成本高收益的优化，Linux 下装上就有 15% 提升，没理由不用。

当然也有代价：异步代码的调试比同步难，栈追踪不直观，团队需要适应。但如果你的服务被 IO 阻塞拖累，这些学习成本绝对值得。

最后留个问题给大家：你们在生产环境用 asyncio 时，有没有遇到过协程泄漏或者事件循环被阻塞的情况？欢迎评论区交流。

本文代码已脱敏，下游 URL 和端口为示例。压测数据基于内网 mock 服务，真实环境会受网络和下游稳定性影响，仅供参考。