1. 问题背景：P99延迟突破900ms，用户开始投诉

上个月我们内部的一个数据查询服务频繁出现超时告警。这个服务负责从多个第三方数据源拉取实时行情，然后聚合返回给前端。业务逻辑不复杂，但每次请求需要串行调用3-5个外部HTTP接口。

先看当时的生产配置：

  • 云服务器：4核8G
  • Python 3.11.4
  • Flask 2.2.5 + Gunicorn 20.1.0（sync worker，4进程）
  • 外部API调用使用 requests 库

压测数据（wrk -t4 -c100 -d30s）：

Requests/sec: 320
Latency Distribution:
  50% 480ms
  75% 620ms
  90% 780ms
  99% 890ms

最让人难受的是：4个 worker 进程的 CPU 总占用率只有38%。显然瓶颈不在计算，而在线程阻塞等待IO。

2. 环境与版本：都说异步好，但别急着上FastAPI

我第一反应是换成 FastAPI + uvicorn，但考虑到底层逻辑和中间件迁移成本，决定先保住 Flask 框架，只把阻塞IO部分改造为异步。

最终测试环境：

Python 3.11.4
Flask 2.2.5
asyncio 3.11.4（内置于标准库）
uvloop 0.19.0
httpx 0.25.1
Gunicorn 20.1.0（uvicorn worker）

为什么用 httpx？因为它原生支持 async/await，且和 requests 的 API 风格接近，改动最小。

3. 方案设计：别全异步，只把IO部分异步化

经验教训：不要试图把整个 Flask 请求处理函数改成 async def。Flask 2.2 的同步视图函数在 async 环境下会有兼容性问题，而且我们还有几个装饰器依赖同步上下文。

我的策略是三层分离：

  1. Web层：保持 Flask 同步视图函数不变
  2. 业务逻辑层：拆分为同步部分和异步部分
  3. IO调用层：用 asyncio.run() 包裹异步函数，或者维护一个全局事件循环

核心思路：在视图函数内部，用 asyncio.run() 执行异步的 IO 任务。虽然这样会为每个请求创建新的事件循环（有开销），但相比阻塞线程，收益大得多。

4. 核心实现：改造前后的代码对比

before：纯同步阻塞

# app.py - 改造前
import requests
from flask import Flask, jsonify

app = Flask(__name__)

EXTERNAL_URLS = [
    "http://data-source-1/api/quote",
    "http://data-source-2/api/quote",
    "http://data-source-3/api/quote",
]

@app.route("/aggregate/")
def aggregate(symbol):
    results = []
    for url in EXTERNAL_URLS:
        # 这里是罪魁祸首：每个请求串行等待3个HTTP响应
        resp = requests.get(f"{url}/{symbol}", timeout=5)
        results.append(resp.json())

    # 模拟聚合逻辑
    return jsonify({
        "symbol": symbol,
        "data": results,
        "total_requests": len(EXTERNAL_URLS)
    })

after：asyncio + httpx 并发请求

# app.py - 改造后
import asyncio
import httpx
from flask import Flask, jsonify

app = Flask(__name__)

EXTERNAL_URLS = [
    "http://data-source-1/api/quote",
    "http://data-source-2/api/quote",
    "http://data-source-3/api/quote",
]

async def fetch_quote(client: httpx.AsyncClient, url: str, symbol: str):
    """异步获取单个外部API数据"""
    try:
        resp = await client.get(f"{url}/{symbol}", timeout=5.0)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        return {"error": str(e), "source": url}

async def gather_quotes(symbol: str):
    """并发拉取所有外部API"""
    # 使用连接池复用TCP连接，避免每次握手
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
        timeout=httpx.Timeout(5.0, connect=2.0)
    ) as client:
        tasks = [fetch_quote(client, url, symbol) for url in EXTERNAL_URLS]
        # 关键：asyncio.gather 并发执行，而不是串行
        return await asyncio.gather(*tasks)

@app.route("/aggregate/")
def aggregate(symbol):
    # asyncio.run 每次创建新的事件循环
    # 注意：Flask 2.2 的 view 函数不能是 async def，所以用 run 包裹
    results = asyncio.run(gather_quotes(symbol))

    return jsonify({
        "symbol": symbol,
        "data": results,
        "total_requests": len(EXTERNAL_URLS)
    })

关键点：

  • asyncio.gather() 把 3 个外部请求从串行变为并发
  • httpx.AsyncClient 内部维护连接池，max_keepalive_connections=20 避免频繁 TCP 握手
  • asyncio.run() 每个请求创建一个事件循环，虽然不完美，但简单可靠

5. 踩坑与优化：asyncio.run 的性能陷阱

改完之后第一轮压测，QPS 从 320 提升到 680，但离预期还差很远。用 py-spy dump 看了一下线程栈，发现瓶颈变成了 asyncio.run() 本身。每次调用它都要创建和销毁事件循环，这个开销在高并发下被放大了。

优化1：全局事件循环

# 在模块顶层创建全局事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

@app.route("/aggregate/")
def aggregate(symbol):
    # 复用全局事件循环，避免重复创建
    results = loop.run_until_complete(gather_quotes(symbol))
    return jsonify({...})

但注意：多 worker 进程下，每个进程都有自己的全局循环，这是安全的。不过 Flask 的 reloader 模式下会报错，部署时关闭 reloader 即可。

优化2：替换 uvloop

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 然后再创建循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

uvloop 是 C 实现的 libuv 绑定，比 asyncio 默认的 selector 循环快约 30%。

优化3：Gunicorn worker 换成 uvicorn

# gunicorn.conf.py
workers = 4  # CPU核心数
worker_class = "uvicorn.workers.UvicornWorker"
bind = "0.0.0.0:8000"
keepalive = 5
timeout = 30

u vicorn worker 天然支持 ASGI，虽然我们用的是 Flask WSGI，但 uvicorn 的 worker 内部有 event loop 管理，比 sync worker 更适合 IO 密集场景。

6. 效果数据：P99延迟下降76%

最终压测结果（wrk -t4 -c100 -d30s）：

指标 改造前 改造后 提升
QPS 320 1210 +278%
P50 延迟 480ms 120ms -75%
P90 延迟 780ms 180ms -77%
P99 延迟 890ms 210ms -76%
CPU 利用率 38% 72% +89%

另外，在外部API响应时间波动时（模拟某数据源慢到2秒），改造前整体请求会卡死（串行等待），改造后其他两个并发请求正常返回，整体延迟只增加约 300ms。

7. 总结：异步不是银弹，但能解决IO密集问题

这次重构的核心收益在于：把串行IO变为并发IO，充分用上了4个CPU核心的网络等待时间。如果你也遇到类似的瓶颈，建议按以下顺序排查：

  1. 先用 py-spy dump 或 perf 确认 CPU 是否真的在等待 IO
  2. 如果 CPU 利用率低、且大量时间在 socket 等待上，考虑异步
  3. 不要盲目全异步，先改造IO密集部分
  4. 注意 asyncio.run() 的创建开销，高并发时一定要复用事件循环
  5. 生产环境一定要关掉 Flask debug/reloader，否则事件循环会冲突

最后，还是那句老话：先测量，再优化。我们花了一周时间改造，但换来了3倍多的吞吐量，值得。