1. 问题背景：同步IO是性能瓶颈，不是Python慢

先说结论：Python慢是个伪命题，真正慢的是阻塞式IO。

我们有个内部订单报表接口/api/orders/summary，逻辑不复杂：先查用户表拿权限范围，再查订单表算金额，最后查商品表做分类汇总。三个查询互相独立，但用的是Flask+SQLAlchemy同步写法：

# 重构前：串行阻塞，三个查询耗时累加
@app.route("/api/orders/summary")
def summary():
    user = db.query(User).filter_by(id=current_user).first()
    orders = db.query(Order).filter(user_id=user.id, status="paid").all()
    products = db.query(Product).filter(ids=[o.product_id for o in orders]).all()
    return jsonify(aggregate(orders, products))

压测数据（wrk -t4 -c100 -d30s）：QPS 120，P95 1850ms，CPU占用率仅35%。很明显，线程在等待数据库IO时全挂起了。Flask默认单线程，虽然我开了threaded=True，但GIL和DB连接池锁依然让并发上不去。

2. 环境与版本：别用旧版本，坑多

我的实测环境（2024年3月）：

  • Python 3.11.8（asyncio在3.10+才有asyncio.timeout，3.12的TaskGroup更香但我没升）
  • Flask 3.0.2（仅保留路由层，不跑WSGI）
  • httpx 0.27.0（异步HTTP客户端，替代requests）
  • uvicorn 0.29.0（ASGI服务器，替代gunicorn）
  • PostgreSQL 15 + asyncpg 0.29.0（异步驱动，比psycopg2快30%）

核心思路：Flask只做路由分发，实际逻辑走asyncio.run()跑协程。虽然这不如直接用FastAPI干净，但业务代码改动最小。

3. 方案设计：三个查询并发，一个Semaphore控流

设计目标：

  • 三个查询并发执行，总耗时≈最慢的那个（原先是三个之和）
  • 防止DB连接池被打爆，用asyncio.Semaphore(10)限制同时查询数
  • 单个查询失败不影响整体，用return_exceptions=True隔离异常
# 重构后：异步并发，三个查询同时飞
import asyncio, httpx
from flask import Flask, jsonify

app = Flask(__name__)
_semaphore = asyncio.Semaphore(10)

async def fetch_user(db_pool, uid):
    async with _semaphore:
        return await db_pool.fetchrow("SELECT * FROM users WHERE id=$1", uid)

async def fetch_orders(db_pool, uid):
    async with _semaphore:
        return await db_pool.fetch("SELECT * FROM orders WHERE user_id=$1 AND status='paid'", uid)

async def fetch_products(db_pool, ids):
    async with _semaphore:
        return await db_pool.fetch("SELECT * FROM products WHERE id = ANY($1)", ids)

async def handler(db_pool, uid):
    # gather并发执行，return_exceptions防止一个挂全部挂
    user, orders, products = await asyncio.gather(
        fetch_user(db_pool, uid),
        fetch_orders(db_pool, uid),
        fetch_products(db_pool, []),  # 先传空，依赖orders结果
        return_exceptions=True
    )
    # 注意：products依赖orders，这里拆两步
    if isinstance(orders, Exception) or isinstance(user, Exception):
        return {"error": "query failed"}, 500
    products = await fetch_products(db_pool, [o["product_id"] for o in orders])
    return aggregate_data(user, orders, products)

@app.route("/api/orders/summary")
def summary():
    uid = current_user_id()
    db_pool = app.config["db_pool"]
    result, status = asyncio.run(handler(db_pool, uid))
    return jsonify(result), status

4. 核心实现：asyncpg连接池 + 两步gather

上面代码有个依赖问题：products查询依赖orders的返回。所以不能简单三个gather，得拆成两轮：

async def handler_optimized(db_pool, uid):
    # 第一轮：并行查用户和订单
    user_task = asyncio.create_task(fetch_user(db_pool, uid))
    orders_task = asyncio.create_task(fetch_orders(db_pool, uid))
    user, orders = await asyncio.gather(user_task, orders_task, return_exceptions=True)

    if isinstance(user, Exception) or isinstance(orders, Exception):
        return {"error": "db error"}, 500

    # 第二轮：根据订单查商品
    product_ids = [o["product_id"] for o in orders]
    products = await fetch_products(db_pool, product_ids)
    return aggregate(user, orders, products)

关键点：asyncio.create_task比gather更灵活，能控制任务粒度。另外asyncpg连接池初始化：

async def init_pool():
    return await asyncpg.create_pool(
        user="postgres", password="xxx", database="orders",
        host="127.0.0.1", port=5432,
        min_size=5, max_size=20,
        command_timeout=5.0  # 超时5秒，防止慢SQL拖死
    )

5. 踩坑与优化：三个血泪教训

坑1：Flask的before_request里不能开事件循环
我最初尝试在before_request里asyncio.run()创建连接池，结果每个请求都新建事件循环，连接池失效。解决：在app.before_serving钩子里创建池，存到app.config。

坑2：asyncio.run()每次调用都销毁重建事件循环
这是性能杀手！我压测时发现QPS只到180，排查发现asyncio.run占用了20%开销。优化：在Flask进程启动时创建一个全局事件循环，用asyncio.get_event_loop()复用：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 在app.before_serving时启动
loop.run_until_complete(init_pool())
# 每个请求用 loop.run_until_complete(handler(...)) 但注意线程安全

更稳妥的方案：直接用uvicorn跑ASGI应用，别用Flask的WSGI。我最后把路由迁到FastAPI，代码几乎没变，性能再涨20%。

坑3：Semaphore的等待时间也算在请求里
当并发超过10，后续协程会排队。压测时P99飙到2s+。优化：把Semaphore从10降到5，减少DB压力，反而因为锁竞争减少，P95更稳定。通过asyncpg的max_size=20和Semaphore联动，让队列在应用层而非DB层。

6. 效果数据：用数字说话

压测环境：4核8G云服务器，PostgreSQL同机部署，wrk压测30秒。

指标 重构前(Flask同步) 重构后(Flask+asyncio) 重构后(FastAPI+asyncio)
QPS 120 248 340
P50/P95/P99 (ms) 650/1850/3200 180/620/1100 120/480/850
CPU占用 35% 55% 68%
DB连接数峰值 25 15 12

提升原因：IO等待时间从三个请求串行的~1.8s压缩到最慢的一个~0.6s。CPU不再是瓶颈，等待IO的时间被并发填满。

7. 总结：异步不是银弹，但IO密集场景必须用

  • 如果你的API是CPU密集（图片处理、加密），asyncio帮不了你，请用多进程
  • 如果是IO密集（查DB、调外部API、读写文件），asyncio是性价比最高的方案
  • 别在Flask里硬塞asyncio，除非你只想快速验证。生产环境直接上FastAPI/Starlette
  • 记得用uvloop替换默认事件循环，再涨10%性能：
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

最后贴一下GAE（Google App Engine）部署时的坑：uvicorn默认单worker，要配--workers 4，但每个worker独立事件循环，Semaphore要按worker数分配。别问我是怎么知道的。