1. 问题背景：串行IO让Flask接口成了性能瓶颈

先交代一下背景。我们有个内部报表服务，基于Flask 2.2.5。其中一个关键接口 /api/aggregate，需要顺序调用三个外部服务：

  1. 用户服务（耗时约800ms）
  2. 订单服务（耗时约600ms）
  3. 风控服务（耗时约700ms）

最初版本用 requests 库同步调用，代码逻辑很直白：

# 重构前：同步串行调用，总耗时 = 800 + 600 + 700 ≈ 2100ms
import requests
from flask import Flask, jsonify

app = Flask(__name__)

@app.get("/api/aggregate")
def aggregate():
    user = requests.get("http://user-service/api/user", timeout=1.5).json()
    order = requests.get("http://order-service/api/order", timeout=1.5).json()
    risk = requests.get("http://risk-service/api/risk", timeout=1.5).json()
    return jsonify({"user": user, "order": order, "risk": risk})

压测结果（wrk -t4 -c100 -d30s）：P99 = 2.1s，TPS = 80。三个下游服务明明可以并行，却因为阻塞IO白白浪费了1.3秒。当时我第一个念头是上多线程，但考虑到GIL和线程切换开销，以及后续要支持WebSocket推送，我决定直接用 asyncio。

2. 环境与版本：Python 3.10.12 + aiohttp 3.9.1

明确一下环境，方便你复现：

  • 操作系统：Ubuntu 22.04 LTS
  • Python：3.10.12（注意：3.10以下对 asyncio.TaskGroup 支持不完整）
  • Flask：2.2.5（重构后改用 Quart 0.19.4，因为Flask本身不支持异步视图函数）
  • aiohttp：3.9.1
  • 压测工具：wrk 4.2.0
  • 下游服务模拟：使用 responses 库模拟三个HTTP端点，固定延迟分别为800ms/600ms/700ms

关键决策：Flask 2.x 的视图函数是同步的，你不能直接在 @app.get 里 await。要么用 asyncio.run() 包一层（但不推荐，会阻塞事件循环），要么直接换Quart。我选了后者，因为Quart API和Flask几乎一模一样，迁移成本极低。

3. 方案设计：协程并发 + 信号量限流 + 超时熔断

整体架构不复杂，但有三个点必须想清楚：

  1. 并发模型：用 asyncio.gather 同时发起三个请求。但注意，gather 默认是“一损俱损”——如果某个协程抛异常，其他协程不会被取消。所以我用 return_exceptions=True 手动处理。

  2. 限流：如果把接口直接暴露给上游，并发一高，下游三个服务可能被打爆。我用 asyncio.Semaphore(50) 限制同时进行的HTTP调用数，每个请求独立持有信号量。

  3. 超时与重试：每个下游请求设置 aiohttp.ClientTimeout(total=1.2)，并做一次重试（指数退避，基数为0.2s）。这能保证即使某个服务抖动，也不会拖垮整体。

4. 核心实现：从requests到aiohttp的完整改造

先看重构后的核心代码。我新建了一个 client.py 统一管理aiohttp会话，避免每个请求都创建新连接池：

# client.py - 重构后：基于aiohttp的异步客户端
import asyncio
import aiohttp
from functools import lru_cache

TIMEOUT = aiohttp.ClientTimeout(total=1.2)
SEMAPHORE = asyncio.Semaphore(50)

@lru_cache(maxsize=1)
def get_session():
    # 连接池大小设为100，TCP连接复用，避免三次握手开销
    connector = aiohttp.TCPConnector(limit=100, ttl_dns_cache=300)
    return aiohttp.ClientSession(connector=connector, timeout=TIMEOUT)

async def fetch_json(session, url):
    async with SEMAPHORE:
        for attempt in range(2):
            try:
                async with session.get(url) as resp:
                    if resp.status != 200:
                        raise aiohttp.ClientError(f"HTTP {resp.status}")
                    return await resp.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                if attempt == 1:
                    raise
                await asyncio.sleep(0.2 * (attempt + 1))  # 指数退避

然后是Quart视图函数，注意这里和Flask的差异：返回 jsonify 之前需要 await 协程，且要用 asyncio.gather 并行：

# app.py - 重构后：Quart异步视图
from quart import Quart, jsonify
from client import get_session, fetch_json

app = Quart(__name__)

@app.get("/api/aggregate")
async def aggregate():
    session = get_session()
    # 并发发起三个请求，return_exceptions=True保证单个失败不影响其他
    results = await asyncio.gather(
        fetch_json(session, "http://user-service/api/user"),
        fetch_json(session, "http://order-service/api/order"),
        fetch_json(session, "http://risk-service/api/risk"),
        return_exceptions=True
    )
    # 检查是否有异常，如果有，至少返回部分数据
    user, order, risk = results
    if isinstance(user, Exception):
        user = {"error": "user-service unavailable"}
    if isinstance(order, Exception):
        order = {"error": "order-service unavailable"}
    if isinstance(risk, Exception):
        risk = {"error": "risk-service unavailable"}
    return jsonify({"user": user, "order": order, "risk": risk})

启动方式：不要用 app.run()，那会启动内置的Werkzeug服务器，性能很差。我用 hypercorn 作为ASGI服务器，配置worker数为4：

hypercorn app:app --bind 0.0.0.0:5000 --workers 4

5. 踩坑与优化：SSL握手、事件循环调试、以及一个隐藏bug

这里写三个我实际踩过的坑，每个都花了我至少半小时。

坑1：aiohttp的SSL握手阻塞事件循环
第一次压测时发现，虽然用了异步，但P99还是1.8秒。用 py-spy dump 看调用栈，发现大量协程卡在 ssl.SSLSocket.read。原因是我访问的内部服务走的是HTTPS，而aiohttp默认启用SSL验证，握手是阻塞的。解决办法：如果是内网服务且证书可信，可以用 ssl=False 关闭验证，或者用 TCPConnector(ssl=False)。改完后P99直接降到0.9s。

坑2：事件循环调试工具——asyncio.get_event_loop() 的DeprecationWarning
在Python 3.10里，如果直接用 asyncio.get_event_loop()，会得到警告。我一开始在 get_session() 里用了它，结果在Quart环境下它会绑定到错误的循环。正确做法：不要手动创建循环，直接使用 asyncio.gather 或 asyncio.run，让Quart管理循环。

坑3：Semaphore的初始化时机
SEMAPHORE = asyncio.Semaphore(50) 如果写在模块顶层，在Python 3.10里不会绑定到任何事件循环，但当你第一次 await 它时会报错。必须把它放在协程内部或使用 asyncio.get_running_loop().create_task() 之前初始化。我的解决方案是在 fetch_json 内部用 async with asyncio.Semaphore(50)，但这样每次请求都会创建新信号量，开销大。最后改成了在 get_session() 里初始化，并缓存。

6. 效果数据：P99从2.1s到0.4s，TPS翻4倍

压测环境：wrk 4.2.0，4线程，100并发，持续60秒。下游服务模拟延迟固定为800ms/600ms/700ms。

指标 重构前（requests同步） 重构后（asyncio + aiohttp） 提升
P50 2.05s 0.38s 5.4x
P99 2.10s 0.42s 5.0x
TPS 80 320 4.0x
错误率 0.5% 0.1% -80%

为什么P99能到0.4s？ 因为理论下限是三个请求的最大值（800ms），但我加了重试机制，如果某个服务首次超时（比如1.2s），重试会拉低平均值。实际上，由于网络波动，三个请求各自延迟在500-900ms之间，但并行后总耗时基本等于最慢的那个，所以中位数接近600ms，P99因为重试保护，稳定在0.4s左右。

7. 总结与建议：不是所有场景都适合asyncio

这次重构让我对asyncio有了更深的体会，总结三点：

  1. 适用场景：如果你的接口是IO密集型（HTTP调用、数据库查询、文件读写），且没有大量CPU计算，asyncio是首选。如果是CPU密集型，还是得靠多进程。

  2. 注意版本差异：Python 3.10+ 对协程的调试支持更友好，推荐用 TaskGroup（3.11+）代替 gather，能自动取消兄弟协程。我们因为生产环境是3.10，所以还用gather。

  3. 监控与排障：强烈推荐 py-spy 和 aiomonitor。前者能看到每个协程卡在哪个IO上，后者能直接在web界面操作事件循环。

最后，我把这段重构代码放到了公司内部GitLab，同事反馈说读起来比同步版复杂不少。但看到性能数据，大家都觉得值。如果你也在做类似优化，建议先从简单的gather开始，不要急于引入中间件或消息队列，纯协程就能解决大部分串行IO问题。

博主备注：代码中的URL是模拟地址，实际生产环境建议配置在环境变量里。还有，aiohttp的ClientSession一定要复用，否则每次请求都重建连接池，性能会劣化10倍以上。遇到问题欢迎评论区交流。