一、问题背景

去年接手一个内部推荐服务，逻辑很简单：客户端请求进来，服务端要调三个下游接口拿数据，然后聚合返回。

  • 用户画像接口：平均耗时 80ms
  • 商品特征接口：平均耗时 120ms
  • 实时热度接口：平均耗时 60ms

三个接口互不依赖，理论上可以并发。但原代码是Flask同步写的，串行调用，一个请求光等下游就要 260ms 起步。加上框架开销和GC，P99直接飙到 480ms。

压测数据（4核8G，8个worker）：

指标 数值
QPS 120
P50 310ms
P99 480ms
机器数 8台

业务方天天抱怨超时，运维说机器不够。我看了一眼代码，问题很明确：IO等待时间占了整个请求的 90% 以上，CPU 基本在摸鱼。

二、环境与版本

Python 3.11.6
Flask 3.0.0
aiohttp 3.9.1
uvicorn 0.25.0
gunicorn 21.2.0
httpx 0.26.0（用于压测对比）

选 Python 3.11 是因为它的 asyncio 在异常处理和 TaskGroup 上比 3.10 快不少，官方 benchmark 显示 async 场景有 10%~20% 的提升。

三、方案设计

核心思路：把 IO 密集的聚合逻辑改成异步，用 asyncio.gather 并发调三个下游。

技术选型上有几个考虑：

  1. 为什么不全换成 FastAPI？ 改造成本太大，接口契约、中间件、鉴权都要动。折中方案是保留 Flask 做路由和参数校验，把耗时的聚合逻辑抽出来用 asyncio 跑。
  2. 怎么在 Flask 里跑 asyncio？ Flask 是 WSGI 同步框架，直接在视图里 asyncio.run() 每次都会新建事件循环，开销大。正确做法是在应用启动时创建一个全局事件循环，用 run_until_complete 或 asyncio.run_coroutine_threadsafe 提交任务。
  3. 下游调用用什么？ aiohttp 的 ClientSession 是连接池复用的关键，绝对不能每次请求新建 session。

四、核心实现

Before：同步串行版本

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

app = Flask(__name__)

def fetch_profile(uid):
    r = requests.get(f"http://profile-svc/user/{uid}", timeout=1.0)
    return r.json()

def fetch_features(uid):
    r = requests.get(f"http://feature-svc/item/{uid}", timeout=1.0)
    return r.json()

def fetch_hotness(uid):
    r = requests.get(f"http://hot-svc/score/{uid}", timeout=1.0)
    return r.json()

@app.route("/recommend/")
def recommend(uid):
    # 串行调用，累加耗时
    profile = fetch_profile(uid)
    features = fetch_features(uid)
    hotness = fetch_hotness(uid)
    return jsonify({
        "uid": uid,
        "profile": profile,
        "features": features,
        "hotness": hotness,
    })

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

这段代码的问题不用多说：三个请求串行，任何一个卡住，后面的都得等。requests 每次新建连接，TCP 握手 + TLS 又是一笔开销。

After：asyncio 并发版本

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

app = Flask(__name__)
LOOP = asyncio.new_event_loop()
SESSION: aiohttp.ClientSession = None

TIMEOUT = aiohttp.ClientTimeout(total=0.8, connect=0.2)
CONNECTOR = aiohttp.TCPConnector(
    limit=200,           # 总连接数上限
    limit_per_host=50,   # 单host连接上限
    ttl_dns_cache=300,   # DNS缓存5分钟
    keepalive_timeout=30,
)

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

async def aggregate(uid):
    # 三个请求并发发出
    results = await asyncio.gather(
        fetch(SESSION, f"http://profile-svc/user/{uid}"),
        fetch(SESSION, f"http://feature-svc/item/{uid}"),
        fetch(SESSION, f"http://hot-svc/score/{uid}"),
        return_exceptions=True,
    )
    profile, features, hotness = results
    # 单个下游失败不影响整体，降级为空
    return {
        "uid": uid,
        "profile": profile if not isinstance(profile, Exception) else {},
        "features": features if not isinstance(features, Exception) else {},
        "hotness": hotness if not isinstance(hotness, Exception) else 0,
    }

@app.route("/recommend/")
def recommend(uid):
    future = asyncio.run_coroutine_threadsafe(aggregate(uid), LOOP)
    return jsonify(future.result(timeout=1.0))

def start_background_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

if __name__ == "__main__":
    import threading
    # 在后台线程跑事件循环，主线程继续跑Flask
    threading.Thread(target=start_background_loop, args=(LOOP,), daemon=True).start()

    # 在事件循环里初始化session，避免跨loop问题
    async def init_session():
        global SESSION
        SESSION = aiohttp.ClientSession(
            connector=CONNECTOR,
            timeout=TIMEOUT,
            headers={"User-Agent": "rec-svc/1.0"},
        )
    asyncio.run_coroutine_threadsafe(init_session(), LOOP).result()

    app.run(host="0.0.0.0", port=8000, threaded=True)

关键点：

  • asyncio.gather 让三个请求真正并发，总耗时取决于最慢的那个（120ms）而不是累加（260ms）
  • TCPConnector 配置连接池，limit_per_host=50 防止单下游被打爆
  • return_exceptions=True 让单个下游失败时整体不挂，配合降级逻辑
  • 全局 SESSION 复用连接，省掉每次 TCP 握手
  • Flask 视图里用 run_coroutine_threadsafe 把协程提交到后台 loop，用 future.result(timeout=1.0) 同步等待结果

五、踩坑与优化

坑1：在Flask视图里直接 asyncio.run()

一开始偷懒，每个请求 asyncio.run(aggregate(uid))，结果压测时 QPS 不升反降。原因是每次调用都新建事件循环、重建 session，连接池完全没复用。改成全局 loop + 全局 session 后 QPS 直接翻倍。

坑2：跨事件循环使用 ClientSession

aiohttp 的 session 绑定创建它的事件循环。我在主线程 asyncio.run 里建 session，在后台 loop 里用，直接报 RuntimeError: Event loop is closed。解决方法是让 session 的创建和使用都在同一个 loop 里。

坑3：连接池打爆下游

limit=200 一开始设得太激进，压测时下游三个服务被我们打挂两次。后来改成 limit_per_host=50，并且和下游团队对齐了限流阈值，稳定了。

坑4：超时设置不合理

total=0.8 是拍脑袋定的。实际观察发现实时热度接口 P99 是 200ms，画像接口 P99 是 180ms，商品特征接口偶尔会飙到 600ms。后来把 total 设成 1.0，connect 设成 0.2，并且给商品特征接口单独配了 1.5s 的超时和熔断降级。

优化：任务组 + 取消传播

Python 3.11 的 TaskGroup 比 gather 更优雅，任何一个子任务抛异常会自动取消其他任务，避免资源泄漏。生产上我最终用的就是 TaskGroup 版本：

async def aggregate_v2(uid):
    result = {"uid": uid, "profile": {}, "features": {}, "hotness": 0}
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch(SESSION, f"http://profile-svc/user/{uid}"))
        t2 = tg.create_task(fetch(SESSION, f"http://feature-svc/item/{uid}"))
        t3 = tg.create_task(fetch(SESSION, f"http://hot-svc/score/{uid}"))
    # 走到这里说明全部成功，任何一个失败会抛ExceptionGroup
    result["profile"] = t1.result()
    result["features"] = t2.result()
    result["hotness"] = t3.result()
    return result

不过 TaskGroup 的异常处理是 ExceptionGroup，需要额外包一层 try/except 做降级，这点不如 gather + return_exceptions 直接。两种都用过，看团队习惯。

六、效果数据

压测工具用 wrk -t4 -c200 -d60s，4核8G机器，gunicorn 4 worker：

指标 Before After 提升
QPS 120 1100 9.2x
P50 310ms 42ms 7.4x
P99 480ms 95ms 5.1x
CPU使用率 25% 68% -
机器数 8台 2台 减少75%

成本核算：8台4核8G换成2台，一年省了小十万服务器费用，运维也轻松了。

七、总结

这次优化核心就一句话：IO 密集场景，别让 CPU 陪着 IO 一起等。

几点经验：

  1. asyncio 不是银弹，CPU 密集场景反而会更慢（GIL + 事件循环开销）。判断标准：IO 等待时间占比 > 70% 才值得改。
  2. 全局事件循环 + 全局 ClientSession 是 Flask 里用 asyncio 的关键，别在视图里 asyncio.run。
  3. 连接池、超时、限流三个参数必须和下游对齐，不然异步化就是把串行压力变成并发压力，更容易打挂下游。
  4. Python 3.11 的 TaskGroup 值得用，但异常处理要额外注意。
  5. 如果项目从零开始，直接上 FastAPI + uvicorn 更省事，不用折腾 WSGI 里嵌 asyncio 这套。

最后提醒一句：异步化改造前先做压测基线，改造后再压一次，数据不会骗人。我见过太多人改完觉得"应该快了"，结果因为连接池没配好反而更慢的案例。