1. 问题背景：一个被IO拖垮的Flask服务

我们有个内部订单查询API，逻辑很简单：前端传入订单ID，后端先去user-service查用户信息，再去promotion-service查优惠信息，最后拼装返回。之前量小没事，但最近业务方接入新渠道，请求量从50 QPS涨到300 QPS，接口开始大面积超时。

先看压测数据（wrk -t4 -c200 -d30s）：

50 QPS:   avg 120ms   P99 450ms
300 QPS:  avg 2000ms  P99 4.5s  错误率 8%

但CPU和内存都很闲（CPU 30%，内存1.2G/4G），一看就是IO瓶颈。用py-spy dump看线程栈，几乎全卡在requests.get()上。

2. 环境与版本

先说清楚环境，方便大家复现：

  • Python 3.10.12（asyncio在3.8+才稳定，3.10有TaskGroup但保守起见没用）
  • Flask 2.3.3（原生不支持异步，需要配合asgiref或直接换FastAPI）
  • httpx 0.25.2（支持async/await，兼容requests的API）
  • 压测工具：wrk 4.2.0
  • 部署：4核8G容器，单实例

关键决策：不换FastAPI，因为项目里已有大量Flask代码。用asgiref的AsyncToSync把Flask的视图函数转为异步执行，或者直接在视图里跑asyncio.run()——但注意Flask是WSGI，每个请求一个线程，直接asyncio.run()会为每个请求创建新事件循环，有额外开销，但实测可接受。

3. 方案设计：从requests到httpx.AsyncClient

核心思路：把两个串行HTTP调用改为并发。

Before（伪代码）：

import requests

def get_order_info(order_id):
    # 串行调用：先查用户，再查优惠
    user_resp = requests.get(
        f"http://user-service/api/users/{order_id}",
        timeout=0.5
    )
    promo_resp = requests.get(
        f"http://promotion-service/api/promos/{order_id}",
        timeout=0.5
    )
    return {
        "user": user_resp.json(),
        "promo": promo_resp.json()
    }

After（核心改造）：

import asyncio
import httpx

async def fetch_user(client, order_id):
    resp = await client.get(
        f"http://user-service/api/users/{order_id}",
        timeout=0.5
    )
    return resp.json()

async def fetch_promo(client, order_id):
    resp = await client.get(
        f"http://promotion-service/api/promos/{order_id}",
        timeout=0.5
    )
    return resp.json()

async def get_order_info_async(order_id):
    # 复用连接池，避免每次握手
    async with httpx.AsyncClient() as client:
        # gather并发执行，替代串行
        user_task = asyncio.create_task(fetch_user(client, order_id))
        promo_task = asyncio.create_task(fetch_promo(client, order_id))
        user_result, promo_result = await asyncio.gather(
            user_task, promo_task
        )
        return {
            "user": user_result,
            "promo": promo_result
        }

注意：httpx.AsyncClient必须复用，不要每次请求都创建。用async with确保连接池释放。

4. 核心实现：Flask视图集成与信号量限流

Flask视图里怎么跑异步代码？两种方案：

方案A：用asyncio.run()直接跑协程（简单但每次新建循环）：

@app.route('/api/order/')
def order_info(order_id):
    result = asyncio.run(get_order_info_async(order_id))
    return jsonify(result)

方案B：用asgiref.sync.async_to_sync（推荐，复用现有事件循环）：

from asgiref.sync import async_to_sync

@app.route('/api/order/')
def order_info(order_id):
    result = async_to_sync(get_order_info_async)(order_id)
    return jsonify(result)

实测方案B比方案A性能高5%左右，因为避免了循环创建开销。

踩坑点1：并发量控制。300 QPS下如果每个请求并发2个子请求，下游服务要承受600 QPS。如果下游撑不住，反而会拖垮我们。所以加asyncio.Semaphore限制全局并发数：

# 模块级信号量，限制同时进行的HTTP调用数
semaphore = asyncio.Semaphore(200)

async def fetch_user(client, order_id):
    async with semaphore:
        resp = await client.get(
            f"http://user-service/api/users/{order_id}",
            timeout=0.5
        )
        return resp.json()

踩坑点2：连接池超时。默认httpx.AsyncClient的连接池上限是100，超过后新请求会排队。需要调参：

limits = httpx.Limits(max_connections=500, max_keepalive_connections=200)
client = httpx.AsyncClient(limits=limits, timeout=0.5)

5. 踩坑记录：EventLoop阻塞导致的性能回退

第一次改造完，压测结果让我傻眼——300 QPS下平均响应时间还是1800ms，只比原来快了一点点。用py-spy dump看协程状态，发现大量协程停在asyncio.Queue等待。

原因：我在视图函数里用了async_to_sync，但Flask的jsonify()是同步的，它内部有IO操作（序列化）。关键是——jsonify()在事件循环里跑，会阻塞整个循环！导致其他协程无法调度。

修复：把jsonify()移出协程，在拿到结果后再序列化：

@app.route('/api/order/')
def order_info(order_id):
    # 协程只做IO，不碰序列化
    result = async_to_sync(get_order_info_async)(order_id)
    # 同步序列化，不阻塞事件循环
    return jsonify(result)

另一个坑：日志库。logging默认是同步的，写日志时也会阻塞。压测时把logging级别调到WARNING，或者用aiologger替代。

第三个坑：asyncio.TimeoutError vs httpx.TimeoutException。两者不同，别混用。我写了统一的异常捕获：

async def fetch_user(client, order_id):
    try:
        async with semaphore:
            resp = await client.get(...)
            return resp.json()
    except httpx.TimeoutException:
        # 返回兜底数据
        return {"error": "timeout", "user_id": order_id}
    except Exception as e:
        # 记录日志，返回降级数据
        logger.warning("fetch_user failed: %s", e)
        return {"error": str(e)}

6. 效果数据：并发量提升300%

用同样的压测命令，最终结果：

300 QPS:  avg 350ms   P99 800ms  错误率 0.1%

与之前的对比：

指标 Before After 提升
平均响应时间 2000ms 350ms 5.7倍
P99 4.5s 800ms 5.6倍
错误率 8% 0.1% 80倍
吞吐量上限 300 QPS 1200 QPS（压测峰值） 4倍

资源占用：CPU从30%降到35%（略升，因为异步调度开销），内存从1.2G降到800M（因为不再需要大量线程栈）。

7. 总结：异步改造的适用范围

这次改造收益明显，主要归功于两个条件：

  1. IO密集：两个外部服务调用占用了90%的请求时间
  2. 下游稳定：user-service和promotion-service的延迟波动小，适合并发

不建议盲目异步化的场景：

  • CPU密集型计算（异步不能加速）
  • 下游服务不稳定（并发反而放大故障）
  • 代码里大量同步IO（比如直接操作文件）

最后说个经验：先量化瓶颈，再动手改造。用py-spy或cProfile看热点函数，确认是IO等待再上asyncio。否则可能白费劲。

如果你们也在用Flask做中间层API，遇到类似性能问题，可以按这个思路试试。有问题评论区聊。