asyncio重构Flask接口，QPS从187到1520的并发改造实录

graph LR
A[客户端] --> B[gunicorn worker线程]
B -->|asyncio.run| C[事件循环]
C --> D[Semaphore(50)] --> E[httpx调用服务A]
C --> F[Semaphore(30)] --> G[httpx调用服务B]
C --> H[Semaphore(80)] --> I[httpx调用服务C]
E & G & I --> J[gather聚合]
J --> B

# before.py
import requests
from flask import Flask, jsonify

app = Flask(__name__)

def fetch_user(order):
    # 模拟外部API，实际耗时200ms
    resp = requests.get(f"http://user-svc/{order['uid']}", timeout=2)
    return resp.json()

def fetch_stock(order):
    resp = requests.get(f"http://stock-svc/{order['sku']}", timeout=2)
    return resp.json()

def fetch_logistics(order):
    resp = requests.get(f"http://logi-svc/{order['oid']}", timeout=2)
    return resp.json()

@app.route('/export')
def export():
    orders = get_orders_from_db()  # 假设已实现
    result = []
    for o in orders[:10]:  # 每页10条
        user = fetch_user(o)      # 200ms
        stock = fetch_stock(o)    # 200ms
        logi = fetch_logistics(o) # 200ms
        result.append({...})
    return jsonify(result)

# after.py (Python 3.11)
import asyncio
import httpx
from flask import Flask, jsonify

app = Flask(__name__)

# 每个下游独立的信号量，防止单点打爆
USER_SEM = asyncio.Semaphore(50)
STOCK_SEM = asyncio.Semaphore(30)
LOGI_SEM = asyncio.Semaphore(80)

async def fetch_user(client, order):
    async with USER_SEM:
        resp = await client.get(f"http://user-svc/{order['uid']}", timeout=2.0)
        resp.raise_for_status()
        return resp.json()

async def fetch_stock(client, order):
    async with STOCK_SEM:
        resp = await client.get(f"http://stock-svc/{order['sku']}", timeout=2.0)
        return resp.json()

async def fetch_logistics(client, order):
    async with LOGI_SEM:
        resp = await client.get(f"http://logi-svc/{order['oid']}", timeout=2.0)
        return resp.json()

async def fetch_order_detail(client, order):
    # 关键点：三个请求并发执行，总耗时约等于最慢的一个
    user_task = asyncio.create_task(fetch_user(client, order))
    stock_task = asyncio.create_task(fetch_stock(client, order))
    logi_task = asyncio.create_task(fetch_logistics(client, order))

    # Python 3.11可以用asyncio.timeout，老版本用wait_for
    try:
        async with asyncio.timeout(3.0):  # 整体超时3秒
            user, stock, logi = await asyncio.gather(
                user_task, stock_task, logi_task
            )
            return merge_data(order, user, stock, logi)
    except asyncio.TimeoutError:
        # 超时任务需要取消，否则会泄漏
        for t in (user_task, stock_task, logi_task):
            t.cancel()
        raise

async def process_batch(orders):
    async with httpx.AsyncClient(http2=True, limits=httpx.Limits(max_connections=200)) as client:
        # 限制并发批次，控内存
        sem = asyncio.Semaphore(20)
        async def wrapped(o):
            async with sem:
                return await fetch_order_detail(client, o)
        return await asyncio.gather(*[wrapped(o) for o in orders])

@app.route('/export')
def export():
    orders = get_orders_from_db()
    # 关键：asyncio.run创建独立事件循环，避免线程间共享循环
    result = asyncio.run(process_batch(orders[:10]))
    return jsonify(result)

# gunicorn.conf.py
workers = 4          # 4个进程，每个进程8线程
threads = 8          # 每个进程8个同步线程
worker_class = 'sync' # 必须用sync，不能用gevent
timeout = 30
keepalive = 5