用asyncio重构Flask接口，QPS从320提升到2100的优化记录

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

app = Flask(__name__)

DOWNSTREAM = {
    "user": "http://user-svc/api/user/{uid}",
    "logistics": "http://logi-svc/api/logistics/{oid}",
    "payment": "http://pay-svc/api/payment/{oid}",
}

@app.route("/api/order/")
def get_order(order_id):
    # 1. 查数据库（同步）
    order = db.query_one("SELECT * FROM orders WHERE id=%s", order_id)
    if not order:
        return jsonify({"error": "not found"}), 404

    # 2. 串行调用三个下游 —— 这就是瓶颈
    user = requests.get(DOWNSTREAM["user"].format(uid=order["user_id"]), timeout=1).json()
    logistics = requests.get(DOWNSTREAM["logistics"].format(oid=order_id), timeout=1).json()
    payment = requests.get(DOWNSTREAM["payment"].format(oid=order_id), timeout=1).json()

    return jsonify({"order": order, "user": user,
                    "logistics": logistics, "payment": payment})

# app_async.py
import asyncio
import httpx
from fastapi import FastAPI, HTTPException
import asyncpg

app = FastAPI()

# 全局共享的httpx客户端，复用连接池
client: httpx.AsyncClient | None = None
pool: asyncpg.Pool | None = None

@app.on_event("startup")
async def startup():
    global client, pool
    # 关键配置：限制连接数，避免打爆下游
    limits = httpx.Limits(
        max_connections=200,
        max_keepalive_connections=50,
        keepalive_expiry=30.0,
    )
    client = httpx.AsyncClient(
        limits=limits,
        timeout=httpx.Timeout(1.0, connect=0.3),
    )
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@db:5432/order",
        min_size=5, max_size=20,
    )

@app.on_event("shutdown")
async def shutdown():
    await client.aclose()
    await pool.close()

async def fetch_json(url: str) -> dict:
    """单个下游调用，失败返回空dict，不抛异常"""
    try:
        resp = await client.get(url)
        resp.raise_for_status()
        return resp.json()
    except (httpx.HTTPError, asyncio.TimeoutError) as e:
        # 记日志，返回降级数据
        return {"_error": str(e)}

@app.get("/api/order/{order_id}")
async def get_order(order_id: int):
    async with pool.acquire() as conn:
        order = await conn.fetchrow(
            "SELECT * FROM orders WHERE id=$1", order_id
        )
    if not order:
        raise HTTPException(status_code=404, detail="not found")

    # 三个下游并发发出
    user_task = fetch_json(f"http://user-svc/api/user/{order['user_id']}")
    logi_task = fetch_json(f"http://logi-svc/api/logistics/{order_id}")
    pay_task = fetch_json(f"http://pay-svc/api/payment/{order_id}")

    user, logistics, payment = await asyncio.gather(
        user_task, logi_task, pay_task,
        return_exceptions=False,  # fetch_json内部已吞异常
    )

    return {"order": dict(order), "user": user,
            "logistics": logistics, "payment": payment}