asyncio重构Flask接口：并发吞吐提升340%的完整记录

# before.py
@app.route('/api/orders', methods=['GET'])
def get_orders():
    user_id = request.args.get('user_id')
    # 1. 查用户订单
    orders = db.session.execute(
        text("SELECT id, product_id, store_id, amount FROM orders WHERE user_id = :uid"),
        {"uid": user_id}
    ).fetchall()
    # 2. 循环查商品和店铺（这里最要命）
    result = []
    for o in orders:
        product = db.session.execute(
            text("SELECT name FROM products WHERE id = :pid"),
            {"pid": o.product_id}
        ).fetchone()
        store = db.session.execute(
            text("SELECT name FROM stores WHERE id = :sid"),
            {"sid": o.store_id}
        ).fetchone()
        result.append({
            "order_id": o.id,
            "product_name": product[0] if product else "N/A",
            "store_name": store[0] if store else "N/A",
            "amount": str(o.amount)
        })
    return jsonify(result)

# async_pool.py
import asyncio
import aiomysql

POOL = None

async def init_pool():
    global POOL
    POOL = await aiomysql.create_pool(
        host='127.0.0.1',
        port=3306,
        user='app_user',
        password='secret',
        db='shop',
        minsize=5,
        maxsize=20,
        autocommit=True,
        echo=False
    )

def get_pool():
    if POOL is None:
        raise RuntimeError("Pool not initialized")
    return POOL

# after.py
import asyncio
import aiomysql
from flask import jsonify, request
from async_pool import get_pool

async def fetch_one(sql, args):
    pool = get_pool()
    async with pool.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute(sql, args)
            row = await cur.fetchone()
            return row

async def fetch_order_details(order):
    """并发查询商品和店铺信息"""
    product_sql = "SELECT name FROM products WHERE id = %s"
    store_sql = "SELECT name FROM stores WHERE id = %s"
    # 关键：gather并发执行两个查询
    product, store = await asyncio.gather(
        fetch_one(product_sql, (order['product_id'],)),
        fetch_one(store_sql, (order['store_id'],))
    )
    return {
        'order_id': order['id'],
        'product_name': product['name'] if product else 'N/A',
        'store_name': store['name'] if store else 'N/A',
        'amount': str(order['amount'])
    }

@app.route('/api/orders', methods=['GET'])
async def get_orders_async():
    user_id = request.args.get('user_id')
    pool = get_pool()
    async with pool.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute(
                "SELECT id, product_id, store_id, amount FROM orders WHERE user_id = %s",
                (user_id,)
            )
            orders = await cur.fetchall()

    if not orders:
        return jsonify([])

    # 并发处理所有订单的详情查询
    tasks = [fetch_order_details(o) for o in orders]
    results = await asyncio.gather(*tasks)
    return jsonify(results)

# 启动时初始化连接池
if __name__ == '__main__':
    asyncio.run(init_pool())
    app.run(host='0.0.0.0', port=5000, threaded=False)  # 注意：关掉线程模式

from asyncio import Semaphore
# 全局信号量，限制最多10个并发查询
sem = Semaphore(10)

async def fetch_one(sql, args):
    async with sem:  # 加锁
        pool = get_pool()
        async with pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                await cur.execute(sql, args)
                return await cur.fetchone()