asyncio重构Flask接口：QPS从327到1892的异步改造实录

Python: 3.10.11 (需要3.8+才能完整使用asyncio.run)
Web框架: Flask 2.2.5 (同步，通过gunicorn运行)
HTTP客户端: requests 2.28.2 (同步)
数据库驱动: psycopg2-binary 2.9.6 (同步)
压测工具: wrk 4.2.0
服务器: 4C8G 阿里云ECS，PostgreSQL 14 (max_connections=200)

gunicorn配置: gunicorn -w 8 -k gthread --threads 16 app:app

同步版本: 请求进来 -> 线程池分配线程 -> psycopg2阻塞查询 -> for循环requests阻塞调用ERP -> 返回
异步版本: 请求进来 -> 主线程asyncio.run -> asyncpg异步查询 -> gather并发调用ERP -> 返回

# app_sync.py  — 改造前代码
from flask import Flask, jsonify, request
import psycopg2
import requests
from psycopg2.pool import ThreadedConnectionPool

app = Flask(__name__)
pool = ThreadedConnectionPool(5, 20, host='localhost', dbname='orders')

@app.route('/api/v1/orders/batch', methods=['GET'])
def batch_query():
    order_ids = request.args.get('ids', '').split(',')
    if len(order_ids) > 50:
        return jsonify({'error': 'max 50 ids'}), 400

    conn = pool.getconn()
    try:
        # 1. 查询主表
        cur = conn.cursor()
        cur.execute("SELECT id, amount FROM orders WHERE id = ANY(%s)", (order_ids,))
        rows = cur.fetchall()
    finally:
        pool.putconn(conn)

    # 2. 串行调用ERP — 这里是最痛点
    erp_results = []
    for oid in order_ids:
        resp = requests.get(f'http://erp.example.com/status/{oid}', timeout=3)
        erp_results.append(resp.json())

    return jsonify({'orders': rows, 'erp': erp_results})

if __name__ == '__main__':
    app.run(threaded=True)

# app_async.py — 改造后核心代码
import asyncio
import uvloop
from flask import Flask, jsonify, request
import httpx
import asyncpg

app = Flask(__name__)
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 全局初始化连接池 (在app启动时调用)
async def init_db_pool():
    return await asyncpg.create_pool(
        host='localhost', database='orders',
        min_size=5, max_size=10,  # 关键：限制连接数，避免打爆数据库
        command_timeout=5
    )

async def fetch_erp_status(client: httpx.AsyncClient, oid: str, sem: asyncio.Semaphore):
    """并发获取ERP状态，使用信号量限制并发数"""
    async with sem:  # 限制ERP并发不超过20
        try:
            resp = await client.get(f'http://erp.example.com/status/{oid}', timeout=3)
            return {'oid': oid, 'status': resp.json().get('status')}
        except httpx.TimeoutException:
            return {'oid': oid, 'status': 'timeout'}

async def handle_batch(order_ids):
    # 数据库异步查询
    pool = await init_db_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT id, amount FROM orders WHERE id = ANY($1)", order_ids
        )

    # 并发ERP调用 — 核心优化点
    async with httpx.AsyncClient(timeout=3) as client:
        sem = asyncio.Semaphore(20)  # 防止打爆ERP
        tasks = [fetch_erp_status(client, oid, sem) for oid in order_ids]
        erp_results = await asyncio.gather(*tasks)

    return {'orders': [dict(r) for r in rows], 'erp': erp_results}

@app.route('/api/v1/orders/batch', methods=['GET'])
def batch_query():
    order_ids = request.args.get('ids', '').split(',')
    if len(order_ids) > 50:
        return jsonify({'error': 'max 50 ids'}), 400

    # Flask同步视图内运行事件循环
    result = asyncio.run(handle_batch(order_ids))
    return jsonify(result)

# 重要：需要关闭已有的asyncpg pool，避免连接泄漏
# 生产用uvicorn或hypercorn跑ASGI，这里为演示保留Flask

# 正确做法 —— 在app.py顶层：
_pool = None

async def get_pool():
    global _pool
    if _pool is None:
        _pool = await asyncpg.create_pool(min_size=5, max_size=10)
    return _pool