asyncio重构Flask接口，QPS从87到312的完整记录

一、问题背景：一个慢得被投诉的报表接口

上个月运维同事转来一条投诉：某个B端客户的报表页面加载要3秒多，客户直接截图发到对接群。我拉了下日志，发现是/api/v1/report/summary这个接口——它要依次调用三个内部服务：

1. 订单服务：获取近30天订单量（平均耗时150ms）
2. 库存服务：获取SKU库存水位（平均耗时180ms）
3. 用户服务：获取用户等级和折扣（平均耗时90ms）

三个服务串行调用，加上框架开销和数据库查询，单请求平均耗时420ms。更糟的是，我们用的是Flask 2.0.3 + Gunicorn（4 worker，每个worker单线程），当并发上来后，每个请求都要排队。压测数据惨不忍睹：

核心问题就一句话：三个HTTP调用是串行的，但彼此之间没有数据依赖。这是典型的IO密集型场景，正好是asyncio的用武之地。

二、环境与版本：先交代清楚再动手

• Python 3.9.7（生产环境是3.9，没法用3.10+的match语法和更优的asyncio API）
• Flask 2.0.3（框架不换，只改内部实现）
• Gunicorn 20.1.0（worker_class保持gthread，线程数调大）
• aiohttp 3.8.1（替代requests发HTTP）
• 压测工具：Apache Bench 2.3 + 自写Python并发脚本

关键决策：不引入FastAPI或Quart，因为路由、中间件、鉴权逻辑都是现成的Flask代码，重写成本太高。我们的方案是在Flask视图函数内部用asyncio.run()驱动协程，虽然这有性能损耗（每个请求创建新事件循环），但实测损耗约5%，可接受。

三、方案设计：三种方案对比后选了“半异步”

我们讨论过三个方案：

1. 全异步框架（Quart）：风险高，所有视图函数都要改，鉴权中间件重写，测试成本大。
2. 线程池并发（concurrent.futures）：用ThreadPoolExecutor(max_workers=10)把三个requests调用丢进去。实现简单，但线程切换开销在GIL下并不比asyncio省，而且每个线程的requests库底层还是同步socket，阻塞时白白占着线程。
3. asyncio + aiohttp（采用）：只改视图函数内部，把三次HTTP调用改为asyncio.gather()并发。Flask本身保持同步，但IO等待时间被压缩到最短。

最终选择方案3，理由：
- 改动范围小（只动一个视图函数）
- 并发模型更高效（单线程事件循环，无GIL争抢）
- aiohttp支持连接池复用，能进一步减少TCP握手开销

四、核心实现：before/after代码对比

4.1 Before：串行requests调用（罪魁祸首）

# app/views/report.py
import requests
from flask import jsonify, request

def get_report_summary():
    user_id = request.args.get('user_id')

    # 串行调用，耗时 = 150 + 180 + 90 = 420ms
    order_resp = requests.get(
        f'http://order-service/api/orders/count',
        params={'user_id': user_id, 'days': 30},
        timeout=2
    )
    order_count = order_resp.json()['count']

    stock_resp = requests.get(
        f'http://stock-service/api/stock/level',
        params={'user_id': user_id},
        timeout=2
    )
    stock_level = stock_resp.json()['level']

    user_resp = requests.get(
        f'http://user-service/api/user/level',
        params={'user_id': user_id},
        timeout=2
    )
    user_level = user_resp.json()['level']

    return jsonify({
        'order_count': order_count,
        'stock_level': stock_level,
        'user_level': user_level,
        'total_time': 'computed_in_middleware'
    })

4.2 After：asyncio并发调用（改造后）

# app/views/report.py
import asyncio
import aiohttp
from flask import jsonify, request

# 全局连接池，复用TCP连接
CONNECTOR = aiohttp.TCPConnector(
    limit=100,          # 连接池大小
    ttl_dns_cache=300,  # DNS缓存5分钟
    force_close=False,  # 保持keep-alive
)

async def fetch_json(session, url, params):
    """统一请求函数，带超时和异常处理"""
    async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=2)) as resp:
        if resp.status != 200:
            return None
        return await resp.json()

async def fetch_all(user_id):
    """并发拉取三个服务数据"""
    # 每个协程内创建session，避免全局session的线程安全问题
    async with aiohttp.ClientSession(connector=CONNECTOR) as session:
        tasks = [
            fetch_json(
                session,
                'http://order-service/api/orders/count',
                {'user_id': user_id, 'days': 30}
            ),
            fetch_json(
                session,
                'http://stock-service/api/stock/level',
                {'user_id': user_id}
            ),
            fetch_json(
                session,
                'http://user-service/api/user/level',
                {'user_id': user_id}
            ),
        ]
        # gather并发执行，return_exceptions=True防止一个失败拖垮全部
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

def get_report_summary():
    user_id = request.args.get('user_id')

    # 每个请求创建新事件循环（Flask同步环境下）
    results = asyncio.run(fetch_all(user_id))

    # 解析结果，处理异常情况
    order_count = results[0].get('count') if isinstance(results[0], dict) else 0
    stock_level = results[1].get('level') if isinstance(results[1], dict) else 'unknown'
    user_level = results[2].get('level') if isinstance(results[2], dict) else 'normal'

    return jsonify({
        'order_count': order_count,
        'stock_level': stock_level,
        'user_level': user_level,
        'async_version': True
    })

关键点说明：

• asyncio.run() 每次创建新事件循环，虽然有点浪费，但避免了在Flask工作线程中管理事件循环的复杂性。
• aiohttp.TCPConnector 定义在模块级别，连接池在多个请求间复用。这是性能提升的重要一环——三个服务总共只建立约3个TCP连接，而不是每个请求3个。
• return_exceptions=True 让gather不会因为一个服务超时而中断另外两个，我们后续再单独处理异常。

五、踩坑与优化：三个坑，每个都真实发生

坑1：asyncio.run() 在Gunicorn多worker下偶发事件循环冲突

现象：部署后偶发RuntimeError: Event loop is closed。

原因：Gunicorn的gthread worker每个线程会复用，asyncio.run()内部会调用loop.close()。如果上一个请求的事件循环还没完全释放，下一个请求就尝试创建新循环，会冲突。

解决：改用asyncio.new_event_loop() + set_event_loop + run_until_complete + close的显式管理方式：

def get_report_summary():
    user_id = request.args.get('user_id')
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        results = loop.run_until_complete(fetch_all(user_id))
    finally:
        loop.close()

坑2：aiohttp连接池耗尽导致雪崩

现象：压测200并发时，报OSError: [Errno 24] Too many open files。

原因：默认TCPConnector(limit=100)，但每个连接占用一个文件描述符。Gunicorn的limit_nofile默认1024，4个worker就是4096。但我们的CONNECTOR是模块级共享的，多个worker进程各自持有自己的连接池，理论上没问题。实际查日志发现是连接池中的连接未正确释放——因为ClientSession每次请求都新建，但连接器是复用的，如果session没有被正确close，连接就不会归还池子。

解决：确保ClientSession使用async with正确关闭，并增加limit到200，同时调大Gunicorn的--limit-no-file 65535。

坑3：DNS解析成为新瓶颈

现象：改造后P95降到215ms，但压测峰值时偶尔出现300ms以上的尖刺。

原因：每次session.get都会解析服务名（如order-service），虽然aiohttp有DNS缓存，但默认缓存时间较短。我们的服务名是K8s内部域名，解析本身要20-30ms。

解决：在TCPConnector中设置ttl_dns_cache=300（5分钟），并提前预热DNS：

# 启动时预热
async def warmup():
    async with aiohttp.ClientSession(connector=CONNECTOR) as session:
        await session.get('http://order-service/health')

asyncio.run(warmup())

六、效果数据：用数字说话

压测环境：4核8G云主机，Gunicorn 4 worker（gthread，threads=8），100并发持续5分钟。

资源占用对比

• CPU：改造前压测时CPU平均75%（Gunicorn worker都在阻塞等待IO，但GIL导致切换频繁）；改造后CPU平均45%（事件循环在IO等待时释放GIL，真正占用CPU的是JSON解析）。
• 内存：改造前每个请求约3个线程栈（每个线程默认8M栈），100并发时内存峰值1.2GB；改造后单线程事件循环，内存峰值650MB。
• TCP连接数：改造前100并发时，每个worker同时持有约25个连接到order-service（峰值100个）；改造后总连接数稳定在6-8个（连接池复用）。

额外收益

• 因为连接池复用，改造后对下游服务的压力也减小了——order-service的负载从85%降到40%。
• 超时错误几乎消失，因为aiohttp.ClientTimeout(total=2)严格生效，而requests的timeout=2在Gunicorn线程调度下经常形同虚设。

七、总结：什么时候该用asyncio？

这次改造的结论很明确：如果你的服务是IO密集型（HTTP调用、数据库查询、文件读写），且这些IO操作相互独立，asyncio能带来3-5倍的吞吐提升。但要注意：

1. 不要盲目全异步：如果你的业务逻辑中有大量CPU计算（如复杂排序、加密解密），asyncio帮不上忙，反而因为事件循环的单线程特性让CPU计算阻塞其他IO任务。
2. 连接池一定要复用：asyncio的优势一半在IO并发，另一半在连接复用。如果每次请求都新建TCP连接，性能提升会大打折扣。
3. 监控事件循环延迟：建议在中间件中记录loop.time()的漂移，如果发现事件循环被阻塞超过50ms，说明有CPU密集任务混进来了。

最后说点个人感受：这次改造最大的收获不是性能数字，而是理解了异步的本质是调度，而不是魔法。asyncio没有让HTTP请求变快，它只是让CPU在等待网络响应时去干别的活。如果你的代码里充满了requests.get()这种同步阻塞调用，再好的异步框架也救不了你。

附录：完整代码已上传到公司内网GitLab，路径/backend/report-service，分支feature/asyncio-optimization。有权限的同事可以直接拉取对比。

（本文所有性能数据在内部压测环境实测得出，生产环境因网络拓扑差异可能有波动，但提升趋势一致。）