asyncio重构Flask接口：QPS从120到1800的并发改造实录

def export_orders(order_ids):
    results = []
    for oid in order_ids:
        # 每个请求等待2秒（第三方接口P99）
        trace = requests.post(LOGISTICS_URL, json={"oid": oid}, timeout=3)
        results.append(parse(trace))
    return results

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/api/export', methods=['POST'])
def export():
    order_ids = request.json['order_ids']
    results = []
    for oid in order_ids:
        # 同步阻塞，每次耗时约2秒
        resp = requests.post(
            "https://logistics.example.com/trace",
            json={"order_id": oid},
            timeout=3
        )
        results.append(resp.json()["data"])
    return jsonify({"data": results})

import asyncio
import httpx
from flask import Flask, request, jsonify

app = Flask(__name__)

# 全局httpx异步客户端，连接池复用
client = httpx.AsyncClient(
    base_url="https://logistics.example.com",
    timeout=httpx.Timeout(3.0, connect=1.0),
    limits=httpx.Limits(max_connections=50, max_keepalive_connections=20)
)

# 信号量：控制第三方API并发数，避免触发限流
semaphore = asyncio.Semaphore(30)

async def fetch_trace(order_id: str) -> dict:
    """异步获取单个订单轨迹，带信号量保护"""
    async with semaphore:
        payload = {"order_id": order_id}
        resp = await client.post("/trace", json=payload)
        resp.raise_for_status()
        data = resp.json()
        return {"order_id": order_id, "trace": data["data"]}

async def async_main(order_ids: list) -> list:
    """并发聚合所有订单数据"""
    tasks = [fetch_trace(oid) for oid in order_ids]
    # gather默认返回顺序与输入一致，ensure_async包装
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 过滤异常，记录错误日志
    valid_results = []
    for r in results:
        if isinstance(r, Exception):
            # 生产环境应记录到日志中心，这里简略处理
            valid_results.append({"error": str(r)})
        else:
            valid_results.append(r)
    return valid_results

@app.route('/api/export', methods=['POST'])
def export_async():
    order_ids = request.json['order_ids']
    if not order_ids:
        return jsonify({"error": "empty list"}), 400

    # 关键：在同步Flask线程中创建独立事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        results = loop.run_until_complete(async_main(order_ids))
    finally:
        loop.close()
        # 注意：不能set_event_loop(None)，否则同线程后续请求会报错

    return jsonify({"data": results, "total": len(results)})

if __name__ == "__main__":
    app.run(threaded=True, processes=1)