asyncio重构Flask接口：QPS从387到2106的完整调优记录

一、问题背景：一个把requests用到极致慢的接口

先说结论：同步IO是原罪。

我们有一个内部运营后台，前端调/api/summary获取当日销售汇总。这个接口内部需要并行调用3个下游服务（订单服务、库存服务、用户服务），每个下游平均耗时800ms。最初实现是标准的Flask + requests：

# 优化前：同步串行调用
@app.route('/api/summary')
def summary():
    order_data = requests.get('http://order-service/api/orders', timeout=2).json()
    stock_data = requests.get('http://stock-service/api/stocks', timeout=2).json()
    user_data = requests.get('http://user-service/api/users', timeout=2).json()
    return jsonify(merge(order_data, stock_data, user_data))

高峰期单机QPS 387，P99延迟3.2s——3个下游串行，最坏情况耗时就是2+2+2=6s，而Flask自带的werkzeug是同步server，每个请求占用一个线程，线程池一满，请求全部排队。

不要提grequests或concurrent.futures.ThreadPoolExecutor，线程切换有开销，而且GIL在IO密集场景下虽然会释放，但线程调度+上下文切换在1000+并发下非常不稳定。asyncio是事件循环驱动，单线程处理万级并发IO，这才是正解。

二、环境与版本说明

先交代环境，避免大家复现时踩版本坑：

Python: 3.10.11（3.8以下不建议，asyncio.run是3.7+才稳定的）
Flask: 2.2.5（注意：Flask 2.x的app.run默认不兼容asyncio，必须用其他server）
gunicorn: 20.1.0（配合uvloop）
uvloop: 0.17.0（替换asyncio默认事件循环，性能提升约15%）
aiohttp: 3.8.4（替代requests）

关键警告：不要试图在Flask的视图函数里直接asyncio.run()跑异步代码。Flask的请求处理是同步的，你asyncio.run()会创建新的事件循环，如果视图函数本身在某个事件循环线程里执行（比如用了flask-sock或web框架集成了async），就会报RuntimeError: asyncio.run() cannot be called from a running event loop。

三、方案设计：协程化改造，但保留Flask

我们的策略是：Flask只做HTTP路由和请求解析，把IO密集的逻辑全部下沉到asyncio协程中，通过asyncio.run()在同步视图里调用。同时，把gunicorn的worker类型从sync改为gevent（兼容asyncio），或者直接用aiohttp作为server——但改动太大，我们选择保留Flask。

核心设计：

1. 全局复用事件循环：asyncio.run()每次会创建新循环，无法复用连接池。改为loop = asyncio.new_event_loop()，用asyncio.set_event_loop(loop)，然后在视图里loop.run_until_complete(main())。
2. aiohttp.ClientSession复用：session内部维护连接池，必须全局单例。
3. 信号量限流：防止下游服务被突发流量打爆，设置Semaphore(100)。

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

4.1 优化前：同步串行（代码见上文）

4.2 优化后：asyncio + aiohttp 并发调用

import asyncio
import aiohttp
import uvloop
from flask import Flask, jsonify
from functools import partial

app = Flask(__name__)

# 全局事件循环和session
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
session = aiohttp.ClientSession(loop=loop)  # 显式绑定循环

# 限流：最多同时100个并发请求
semaphore = asyncio.Semaphore(100)

async def fetch_json(session, url):
    async with semaphore:  # 限流
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=2)) as resp:
                return await resp.json()
        except asyncio.TimeoutError:
            return {'error': 'timeout'}

async def fetch_all():
    # 并发调用3个下游，asyncio.gather并行执行
    order_task = fetch_json(session, 'http://order-service/api/orders')
    stock_task = fetch_json(session, 'http://stock-service/api/stocks')
    user_task = fetch_json(session, 'http://user-service/api/users')
    return await asyncio.gather(order_task, stock_task, user_task)

@app.route('/api/summary')
def summary():
    # 在同步视图里运行协程
    order_data, stock_data, user_data = loop.run_until_complete(fetch_all())
    return jsonify(merge(order_data, stock_data, user_data))

注意几个细节：

• aiohttp.ClientTimeout(total=2)：总超时2秒，和之前requests的timeout=2语义一致，但这里因为是并发，3个请求并行，最坏耗时是2s（不是6s）。
• asyncio.gather：默认是串行await？不，gather是并发调度，三个协程同时创建Task，事件循环交替执行IO等待。
• uvloop：把asyncio的事件循环替换为libuv实现，协程切换更快。如果不装，代码也能跑，但压测数据会差10-15%。

4.3 生产部署：gunicorn + gevent worker

app.run()不支持asyncio，生产环境必须用gunicorn。但gunicorn的sync worker是阻塞的，每个worker同时只能处理一个请求。我们改用gevent worker，gevent本身是协程库，但可以和asyncio共存吗？——有坑。

gevent monkey-patch了time.sleep、socket等标准库，但asyncio的事件循环是纯Python的selectors模块，两者不冲突。但请注意必须在gunicorn启动前初始化全局事件循环，否则每个worker进程会创建独立的循环，session连接池就废了。

gunicorn配置（gunicorn.conf.py）：

workers = 4  # 4个进程
worker_class = 'gevent'  # 每个进程内gevent协程调度
threads = 1
timeout = 30
graceful_timeout = 10
preload_app = True  # 提前初始化全局session和loop，避免重复创建

preload_app = True很重要——这样在worker fork之前，我们的session和loop已经创建好，子进程共享同一个事件循环（fork时复制内存），但注意每个进程的session连接池独立，总共4*100=400个下游连接。

五、踩坑与优化记录

5.1 坑1：asyncio.run()报错

最初我在视图里直接写asyncio.run(fetch_all())，结果压测时偶发报错：

RuntimeError: asyncio.run() cannot be called from a running event loop

原因：gunicorn的gevent worker内部有协程调度，当请求进来时，gevent会创建一个greenlet，这个greenlet里可能已经有事件循环（gevent内部用hub loop）。asyncio.run()会强制创建新循环，冲突。

解决：改为显式loop.run_until_complete()，并提前创建全局loop。

5.2 坑2：数据库/其他同步阻塞调用

如果协程里混入了同步的数据库查询（比如psycopg2），会阻塞整个事件循环。因为asyncio是单线程，一旦有阻塞调用，后续所有协程都卡住。

解决：把数据库操作改为aiopg或asyncpg，或者把同步操作丢到loop.run_in_executor(None, sync_func)线程池执行。我们的场景正好都是HTTP调用，所以没踩这个坑，但如果有同学要改成查MySQL，务必注意。

5.3 优化：连接池参数调优

aiohttp.ClientSession默认连接池是无限，但实际需要限制。我们调了：

connector = aiohttp.TCPConnector(limit=200, limit_per_host=50, ttl_dns_cache=300)
session = aiohttp.ClientSession(connector=connector, loop=loop)

• limit=200：全局最大连接数
• limit_per_host=50：每个下游域名最多50个并发连接
• ttl_dns_cache=300：DNS缓存5分钟，避免重复解析

5.4 优化：超时分级

下游服务不稳定，我们把超时分级：
- 连接超时：1s
- 总超时：2s（保持和原逻辑一致）

六、性能对比数据

压测工具：wrk，环境：4核8G云主机，模拟1000并发，持续30秒。

注意：QPS从387涨到2106，但下游服务只允许100并发，所以限流起了作用。如果不加信号量，QPS可能更高，但下游会挂。

容量对比：优化前4台机器才勉强扛住高峰期，优化后1台机器还有30%余量。直接省了3台ECS。

七、总结与建议

1. asyncio适合高IO并发，不适合CPU密集。我们的场景是3个HTTP调用，完美匹配。
2. 别在Flask里硬塞asyncio，除非你换aiohttp或FastAPI。Flask同步模型和asyncio混用有边界问题，我们是用loop.run_until_complete()做桥接。
3. 连接池和限流是生产环境的生命线，没有限流的asyncio会让下游雪崩。
4. uvloop值得装，零代码改动，15%性能提升。
5. 如果让我重来，我会直接选FastAPI——原生支持async，不需要这些hack。但存量Flask项目，这篇文章的方案就是最小改造。

最后贴一下压测命令，方便大家复现：

wrk -t8 -c1000 -d30s --latency http://localhost:8000/api/summary

一点心里话：网上很多文章吹asyncio各种好，但实际项目里坑比想象多。这篇写的是我们真实踩完坑之后的结果，希望你们少走弯路。有疑问评论区聊。