1. 问题背景：一个慢到被投诉的报表接口

上个月，我们数据分析平台有个/api/v1/report/summary接口被业务方投诉了。这个接口的逻辑很简单：前端需要展示一个汇总卡片，包含用户数、订单量、营收三个指标。但这三个数据分别存储在三个不同的微服务里——用户服务、订单服务、财务服务。

最初的实现是同步串行调用，伪代码如下：

def get_summary():
    users = requests.get("http://user-service/api/count", timeout=2).json()
    orders = requests.get("http://order-service/api/count", timeout=2).json()
    revenue = requests.get("http://finance-service/api/revenue", timeout=2).json()
    return {"users": users, "orders": orders, "revenue": revenue}

三个服务平均响应时间都是700-800ms，串行下来就是2.3秒左右。更糟的是，我们的API网关超时阈值是3秒，高峰期经常直接504。客户反馈“页面转圈转半天”，领导让我必须解决。

2. 环境与版本：先交代清楚再动手

  • Python: 3.10.8（生产环境），本地测试也用了3.11.4做对比
  • Web框架: Flask 2.2.3（同步框架，后面会说怎么兼容asyncio）
  • HTTP客户端: httpx 0.24.1（支持async/await）
  • 部署: Gunicorn 20.1.0，4个worker，每个worker单线程
  • 压测工具: wrk 4.2.0，测试时长30秒，并发50连接
  • 三个下游服务：均部署在同一内网，延迟模拟为正常+抖动（±150ms）

3. 方案设计：asyncio不是银弹，但这里是完美匹配

接口是纯IO密集型——CPU占用几乎为0，所有时间都在等网络响应。这正是asyncio的经典使用场景。

核心思路：
1. 用httpx.AsyncClient替代requests
2. 用asyncio.gather并发发起三个请求
3. 由于Flask是同步WSGI框架，不能直接跑async函数——需要asyncio.run()包一层，或者用asgiref.sync.async_to_sync桥接

我选择了asgiref，因为它在Django社区久经考验，而且能在同一个event loop里复用连接池，避免反复创建开销。

设计要点：
- 每个请求创建一个AsyncClient实例，设置limits=httpx.Limits(max_connections=50)
- 使用asyncio.Semaphore(20)限制并发度，防止下游服务被打爆
- 设置总超时5秒，单个请求超时2秒，避免极端情况拖垮整体

4. 核心实现：改造代码和对比

4.1 同步版本（改造前）

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

app = Flask(__name__)

@app.route("/api/v1/report/summary")
def get_summary():
    try:
        users = requests.get("http://user-service/api/count", timeout=2).json()
        orders = requests.get("http://order-service/api/count", timeout=2).json()
        revenue = requests.get("http://finance-service/api/revenue", timeout=2).json()
        return jsonify({"users": users["count"], "orders": orders["count"], "revenue": revenue["amount"]})
    except requests.Timeout:
        return jsonify({"error": "downstream timeout"}), 504

4.2 异步版本（改造后）

# async_after.py
import asyncio
import httpx
from flask import Flask, jsonify
from asgiref.sync import async_to_sync

app = Flask(__name__)

async def fetch_json(client, url):
    resp = await client.get(url, timeout=2.0)
    resp.raise_for_status()
    return resp.json()

async def fetch_all():
    # 设置连接池和并发信号量
    limits = httpx.Limits(max_connections=50, max_keepalive_connections=20)
    semaphore = asyncio.Semaphore(20)

    async with httpx.AsyncClient(limits=limits) as client:
        async def bounded_fetch(url):
            async with semaphore:
                return await fetch_json(client, url)

        # 并发发起三个请求，asyncio.gather等待全部完成
        results = await asyncio.gather(
            bounded_fetch("http://user-service/api/count"),
            bounded_fetch("http://order-service/api/count"),
            bounded_fetch("http://finance-service/api/revenue"),
            return_exceptions=True  # 防止单个失败导致全部失败
        )

    # 检查是否有异常
    for r in results:
        if isinstance(r, Exception):
            raise r
    return results

@app.route("/api/v1/report/summary")
def get_summary():
    try:
        users, orders, revenue = async_to_sync(fetch_all)()
        return jsonify({
            "users": users["count"],
            "orders": orders["count"],
            "revenue": revenue["amount"]
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 504

注意return_exceptions=True这个参数——如果某个下游挂了，我们不能让整个请求失败，而是要继续等待其他两个成功返回，最后再统一报错。

5. 踩坑与优化：三个真实教训

5.1 坑1：asyncio.run()在Flask里会创建新的event loop，导致连接池失效

我第一版用的是asyncio.run(fetch_all())，结果每个请求都新建event loop，AsyncClient的连接池每次都要重新建立TCP连接。内网环境下TCP握手20ms左右，看似不严重，但压测时发现吞吐量上不去，因为每次请求都有额外的3次TCP建立开销。

解决：改用asgiref.sync.async_to_sync，它在Flask进程启动时创建一个持久化的event loop，所有请求共享同一个loop和连接池。

5.2 坑2：信号量忘了加，压测时把下游打挂了

第一版没有Semaphore，直接gather三个请求。wrk压测50并发，瞬间产生150个并发请求打到下游。下游服务直接CPU飙到95%，P99延迟从800ms变成2秒+。

解决：加asyncio.Semaphore(20)，全局限制最多20个并发请求。信号量要定义在fetch_all外层，这样所有请求共享同一个计数。我最初放在async with里面，导致每个请求创建新的信号量，等于没限制。

5.3 优化：Python 3.11的性能提升

生产环境是3.10，本地测试了3.11。发现3.11的asyncio调度性能提升了约15%，特别是gather在任务数量少的情况下，开销更小。用3.11跑同样的压测，QPS从220提升到252。如果你的业务对延迟敏感，值得升级。

6. 效果数据：用数字说话

压测环境：wrk -t4 -c50 -d30s，请求/api/v1/report/summary

指标 同步版本 异步版本 提升幅度
平均延迟 2.31s 0.42s 82%↓
P99延迟 2.87s 0.63s 78%↓
QPS 45 220 389%↑
超时请求(>3s) 12.4% 0% 彻底解决
CPU占用(worker) 18% 22% 可接受

额外观察：
- 异步版本在50并发下，下游服务的请求峰值从150降到20，下游CPU占用从95%降至55%
- 异步版本的连接复用率非常高，max_keepalive_connections=20足够支撑220 QPS

7. 总结：什么时候该用asyncio

这次改造带来的收益远超预期，但我想说asyncio不是万能的。

适合asyncio的场景：
- 纯IO密集型，CPU消耗极低
- 需要并发调用多个外部服务
- 下游服务本身响应时间不短（至少50ms以上），否则异步开销反而变大

不适合的场景：
- CPU密集型计算（应该用多进程）
- 下游服务延迟极低（比如本地Redis <1ms），异步收益不明显
- 代码里混有阻塞调用（比如同步的requests），会完全卡住event loop

最后说一句：如果你还在用Flask，async_to_sync桥接方案是成本最低的改造方式。你不需要换FastAPI或Sanic，只需要把阻塞的IO调用改成async版本，包一层桥接即可。我们的生产环境已经稳定运行3周，无内存泄漏，无异常重启。

如果你也在做类似的改造，建议先把压测脚本准备好，用数据验证每一步优化。祝改造顺利。