asyncio重构Flask接口：QPS翻3倍与线程池的取舍实录

1. 问题背景：一个用户聚合接口拖垮了网关

上个月运维同事给我发来一张监控截图：/api/v1/user/detail 接口在压测环境下，QPS刚到200，P99延迟直接破1秒，CPU 85%，内存2.1GB。看了下代码，这个接口的逻辑是——先查MySQL拿到用户基础信息，然后同步调用三个下游HTTP服务（订单、积分、优惠券），用requests库逐个请求，最后合并数据返回。

# 优化前：同步requests串行调用
def get_user_detail(user_id):
    user = db.query(User).get(user_id)
    order = requests.get(f"http://order-service/api/orders?uid={user_id}").json()
    point = requests.get(f"http://point-service/api/points?uid={user_id}").json()
    coupon = requests.get(f"http://coupon-service/api/coupons?uid={user_id}").json()
    return {...}

三个下游服务平均响应时间都是200ms左右，串行就是600ms，加上DB查询和序列化，单个请求总耗时约700ms。这在低并发下没毛病，但一旦QPS上去，线程池（Flask默认的threaded=True，线程数不限制）疯狂创建线程，每个线程阻塞在requests的socket等待上，GIL和内存双双爆炸。

2. 环境与版本：Python 3.10 + Flask 2.3 + httpx 0.24

这次重构的基准环境：
- Python 3.10.12（asyncio在3.10已经很稳，3.8的坑太多不推荐）
- Flask 2.3.2（配合asgiref的async_to_sync兼容）
- httpx 0.24.1（支持异步的HTTP客户端，比aiohttp更顺手）
- 压测工具：wrk 4.2.0，单机4核8G，下游服务用gunicorn起的Flask mock（响应200ms固定延迟）

关键一点：我没有直接用asyncio.run()包整个Flask路由，因为Flask是WSGI同步框架，直接跑asyncio会阻塞事件循环。正确姿势是——在同步视图函数里，用asyncio.run()启动一个独立的事件循环，跑完就关闭。

3. 方案设计：asyncio + httpx并发请求下游

核心思路很简单：把三个串行的HTTP调用改成并发协程。但有几个设计决策：

1. 并发模型：用asyncio.gather()同时发起三个请求，每个请求一个协程。
2. 连接池：httpx的AsyncClient内部维护连接池，复用TCP连接，避免反复三次握手。
3. 限流：如果下游服务扛不住，并发会打垮它们。所以加asyncio.Semaphore(10)限制最大10个并发。
4. 超时控制：每个请求设置5秒超时，防止某个下游hang住拖死整个接口。

4. 核心实现：改造后的异步代码

# 优化后：asyncio + httpx并发请求
import asyncio
import httpx
from functools import wraps

# 每个worker进程共用一个连接池
_client = None

def get_client():
    global _client
    if _client is None:
        _client = httpx.AsyncClient(
            timeout=httpx.Timeout(5.0, connect=2.0),
            limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
            headers={"X-Request-ID": "trace-123"}
        )
    return _client

def async_route(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        return asyncio.run(f(*args, **kwargs))
    return wrapper

@async_route
async def get_user_detail_async(user_id):
    # DB查询保持同步（也可以用aiomysql，但这里不是瓶颈）
    user = db.query(User).get(user_id)
    if not user:
        return {"error": "user not found"}, 404

    sem = asyncio.Semaphore(10)
    client = get_client()

    async def fetch(url):
        async with sem:
            resp = await client.get(url)
            return resp.json()

    # 并发发起三个下游请求
    order_task = fetch(f"http://order-service/api/orders?uid={user_id}")
    point_task = fetch(f"http://point-service/api/points?uid={user_id}")
    coupon_task = fetch(f"http://coupon-service/api/coupons?uid={user_id}")

    order_data, point_data, coupon_data = await asyncio.gather(
        order_task, point_task, coupon_task,
        return_exceptions=True  # 防止一个失败导致全部失败
    )

    # 处理可能的异常（降级返回空数据）
    order_data = order_data if isinstance(order_data, dict) else {}
    point_data = point_data if isinstance(point_data, dict) else {}
    coupon_data = coupon_data if isinstance(coupon_data, dict) else {}

    return {"user": user.to_dict(), "orders": order_data, 
            "points": point_data, "coupons": coupon_data}

注意return_exceptions=True这个参数——我在压测时发现，一旦下游某个服务超时，gather()会直接抛异常，导致整个接口500。加上这个参数后，单个服务挂了只影响该服务的数据，其他数据正常返回，降级逻辑更健壮。

5. 踩坑与优化：三个真实教训

坑1：asyncio.run()每次创建新事件循环，连接池失效

最初我把httpx.AsyncClient创建放在async_route装饰器内部，结果每个请求都新建client，连接池完全没复用，性能只比同步好一点点（因为并发请求了，但TCP握手开销巨大）。后来改成模块级单例_client，用get_client()懒加载，压测QPS直接翻了2倍。

坑2：Flask的threaded=True和asyncio.run的线程安全

Flask默认每个请求一个线程，asyncio.run()在线程内创建事件循环没问题。但如果开了processes=2（多进程），每个进程都会创建自己的连接池，内存占用会涨。解决方案：用gunicorn -w 4 -k gthread，每个worker进程一个连接池，别用sync worker（那是单线程的）。

坑3：Semaphore的初始值要测试

我一开始用Semaphore(50)，结果下游服务直接被打满（因为它们也是Flask，线程数不设限）。调低到10后，下游P99稳定在280ms，整体接口P99才210ms。这个值需要根据下游的吞吐能力实测调整，没有银弹。

6. 效果数据：QPS翻3.3倍，P99降78%

压测配置：wrk -t4 -c200 -d30s，每个请求随机user_id（保证DB查询不缓存）。下游mock服务固定延迟200ms，带1%的随机抖动。

值得注意的是，内存下降主要是因为连接池复用，而不是并发本身——同步版每个线程的requests会创建独立的socket连接，200个线程就是200个连接，每个连接约5KB缓冲，再加上线程栈（默认8MB），内存就爆了。异步版只用少量协程，内存开销小得多。

7. 总结：什么时候该用asyncio，什么时候别用

这次重构让我彻底明白了asyncio的适用场景——IO密集型且下游服务延迟稳定。如果你下游服务的延迟波动大（比如5%的请求要10秒），异步等待这些慢请求反而会占用事件循环资源，不如用线程池+超时熔断。

另外，如果项目已经用了FastAPI或Sanic，直接原生async支持，不用走asyncio.run()这个hack。但对于存量Flask项目，这个改造方案侵入性最小——只改视图函数内部逻辑，路由和中间件完全不动，灰度发布也简单（按路由切流量）。

最后，记住一个原则：异步解决的是等待问题，不是计算问题。如果你的接口瓶颈在DB查询或CPU计算，asyncio帮不了你，去换数据库连接池或加缓存吧。