用asyncio重构Flask API，并发性能提升300%的完整记录

一、问题背景：一个让老板半夜打电话的同步阻塞

我们有个内部数据聚合服务，代码逻辑很简单：前端请求进来，后端同步调用3个上游HTTP API（用户信息、订单列表、推荐内容），然后拼装返回。这逻辑没毛病，但坏就坏在用的是Flask原生同步视图 + requests库。

线上单机4核8G，部署了4个Gunicorn worker（gevent模式）。高峰期单worker并发处理能力有限，每个请求占着worker线程去等上游IO，平均等待600ms，导致worker线程池被占满，新请求全部排进队列。

压测数据惨不忍睹（wrk -t4 -c200 -d30s）：
- QPS: 82
- P99: 1830ms
- 错误率: 4.2%（超时）

老板半夜看到监控大屏直接打电话：“这个月底要上活动，你给我搞定。”

二、环境与版本：Python 3.10的福利

先交代环境，方便大家复现：
- Python 3.10.12（注意，3.10才有asyncio.TaskGroup，3.9没有）
- Flask 2.3.3（同步框架，但我们只用它做路由分发）
- aiohttp 3.9.1（异步HTTP客户端）
- Gunicorn 21.2.0（用uvicorn换掉？不，我用gunicorn + aiohttp的ASGI模式更省事，但为了最小改动，我选择在Flask视图内部运行事件循环——等等，这埋了个大坑，后面细说）
- 压测工具：wrk 4.2.0

关键点：我们最终没有把Flask换成FastAPI，因为业务代码几百个路由，全重写不现实。所以我的方案是“在同步框架里跑异步IO”——用asyncio.run()包住每个请求的处理逻辑。

三、方案设计：三种方案对比后我选了“混搭”

我列了三个方案：

方案A：全量迁FastAPI — 风险太大，业务代码改动多，测试回归量大，老板等不及。

方案B：用gunicorn的gevent + requests — 其实线上已经是这组合了，但性能提升有限，因为requests是阻塞IO，gevent能切换协程但requests的socket阻塞会卡住整个线程。

方案C：Flask视图内部用asyncio.run() + aiohttp并发调用上游 — 改动最小，只改视图函数内部实现，路由注册不动。我选了C，但做了个关键优化：把asyncio.run()的结果缓存到loop-level的connector，避免每次创建连接池。

四、核心实现：Before/After代码对比

Before：同步阻塞版（线上故障代码）

# app.py (before)
import requests
from flask import Flask, jsonify

app = Flask(__name__)

UPSTREAM_URLS = [
    "http://user-service/api/user",
    "http://order-service/api/orders",
    "http://rec-service/api/recommend"
]

@app.route("/api/aggregate")
def aggregate():
    results = []
    for url in UPSTREAM_URLS:
        # 每个请求同步阻塞，最慢的一个决定整体耗时
        resp = requests.get(url, timeout=5)
        resp.raise_for_status()
        results.append(resp.json())
    return jsonify({"data": results})

这段代码的痛点：requests.get是阻塞的，3个串行请求，假设每个200ms，总耗时就是600ms。如果其中一个上游慢（比如超时5s），整个请求就挂了。

After：异步并发版（改造后）

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

app = Flask(__name__)

UPSTREAM_URLS = [
    "http://user-service/api/user",
    "http://order-service/api/orders",
    "http://rec-service/api/recommend"
]

# 全局连接池，限制并发连接数，避免耗尽fd
connector = aiohttp.TCPConnector(limit=100, limit_per_host=30, ttl_dns_cache=300)

async def fetch_json(session, url):
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=3)) as resp:
        resp.raise_for_status()
        return await resp.json()

async def fetch_all():
    async with aiohttp.ClientSession(connector=connector) as session:
        # 使用gather并发执行，返回顺序与传入顺序一致
        tasks = [fetch_json(session, url) for url in UPSTREAM_URLS]
        return await asyncio.gather(*tasks, return_exceptions=True)

@app.route("/api/aggregate")
def aggregate():
    # 关键：每个请求创建新的事件循环，避免跨线程共享loop
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        results = loop.run_until_complete(fetch_all())
        # 处理异常：如果某个上游挂了，返回部分数据
        data = []
        for r in results:
            if isinstance(r, Exception):
                data.append({"error": str(r)})
            else:
                data.append(r)
        return jsonify({"data": data})
    finally:
        loop.close()

注意几个细节：
1. connector是全局的，但ClientSession是每次请求新建（因为它绑定loop）。connector复用了底层连接池，避免每次握手。
2. loop = asyncio.new_event_loop() 而不是 asyncio.run()，因为我要控制loop生命周期，并在finally里关闭。
3. return_exceptions=True 避免一个上游挂了导致整个gather失败。

五、踩坑与优化：三个我花了一整天才解决的大坑

坑1：事件循环被线程池阻塞（最隐蔽）

我一开始用asyncio.run()放在视图函数里，压测发现QPS反而降到50。后来用py-spy dump查看线程栈，发现每个请求都在等loop.run_until_complete，但事件循环根本没在跑——因为Gunicorn的worker线程池里，多个请求共享同一个asyncio.run()创建的循环？不，asyncio.run()每次创建新loop，但问题在于aiohttp的ClientSession如果绑定到某个loop，跨loop使用会抛异常。我全局创建了ClientSession，它内部绑定到创建它的loop，而每个请求是不同的loop，导致异常被吞或连接复用失效。

解决：必须把ClientSession放进每个loop内创建，但connector可以复用（因为connector内部是独立的传输层连接池，不绑定loop）。这就是我上面代码注释里的写法。

坑2：TCP连接数耗尽

压测时发现大量Connection reset by peer，检查发现aiohttp默认连接池无限大，但系统fd限制1024。我用TCPConnector(limit=100)限制总连接数，同时设置limit_per_host=30避免对单个上游打满连接。

坑3：Flask同步视图的“假异步”陷阱

Flask视图是同步函数，我虽然内部用了asyncio，但Gunicorn的worker线程池大小默认是(2*CPU)+1，即9个线程。每个线程同时只能处理一个请求，所以并发上限还是9。真正提升性能是因为每个请求的耗时从600ms降到150ms（并发3个上游），所以QPS = 9线程/0.15s = 60？不对，实际上Gunicorn的gevent worker模式下，线程是协程，但Flask视图是同步的，gevent能切换，但aiohttp的loop是独立的，所以gevent的monkey patch对asyncio无效。

最终我用了gunicorn -k gevent，但把worker_connections调大（默认1000），并且视图函数内用asyncio时，gevent的协程调度不会干扰asyncio，因为事件循环是独立运行的。实测下来，gevent worker + asyncio内部并发，是最优解。

六、效果数据：用wrk压测的对比

压测环境：4核8G云主机，上游服务用python -m http.server模拟（每个响应延迟200ms）。

命令：wrk -t4 -c200 -d30s http://localhost:8080/api/aggregate

额外调优：把TCPConnector的limit从100调到200，QPS还能再涨到350，但fd占用高，线上我保守用了100。

七、总结：异步不是银弹，但这次值了

这次改造的核心收益不是“用了asyncio就快”，而是把串行阻塞IO变成并发非阻塞IO。如果你的瓶颈是CPU密集，异步毫无帮助；但如果瓶颈是网络IO等待，异步+并发是性价比最高的方案。

最后提醒几点：
- 不要用asyncio.run()在Flask视图里反复创建loop，性能损耗大；用new_event_loop + close。
- 连接池一定要限制大小，否则fd耗尽比超时更可怕。
- 监控用py-spy dump定位线程阻塞，比猜快得多。

如果你也在Flask+Django这类同步框架里做异步改造，希望这篇能帮你少踩一些坑。有问题评论区聊，我基本都在。