用asyncio重构订单查询API：QPS从120提升到1800的调优记录

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

app = Flask(__name__)

def get_conn():
    return psycopg2.connect(
        host="127.0.0.1", port=5432, dbname="orderdb",
        user="app", password="app", connect_timeout=3
    )

@app.route("/api/order/detail")
def order_detail():
    order_id = request.args.get("order_id")
    conn = get_conn()
    cur = conn.cursor()
    try:
        # 1. 主表
        cur.execute("SELECT id, user_id, amount, status FROM orders WHERE id=%s", (order_id,))
        row = cur.fetchone()
        if not row:
            return jsonify({"code": 404, "msg": "not found"}), 404
        order = {"id": row[0], "user_id": row[1], "amount": float(row[2]), "status": row[3]}

        # 2. 商品明细
        cur.execute("SELECT sku, qty, price FROM order_items WHERE order_id=%s", (order_id,))
        order["items"] = [{"sku": r[0], "qty": r[1], "price": float(r[2])} for r in cur.fetchall()]

        # 3. 支付流水
        cur.execute("SELECT pay_no, paid_at FROM payments WHERE order_id=%s", (order_id,))
        order["payments"] = [{"pay_no": r[0], "paid_at": str(r[1])} for r in cur.fetchall()]

        # 4. 物流（第三方HTTP，同步阻塞）
        resp = requests.get(f"http://logistics.internal/api/status?order_id={order_id}", timeout=2)
        order["logistics"] = resp.json() if resp.ok else {}

        return jsonify({"code": 0, "data": order})
    finally:
        cur.close()
        conn.close()

# app_async.py —— 改造后
import asyncio
import asyncpg
import aiohttp
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager

app = FastAPI()
pool: asyncpg.Pool = None
http_session: aiohttp.ClientSession = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool, http_session
    pool = await asyncpg.create_pool(
        host="127.0.0.1", port=5432, database="orderdb",
        user="app", password="app",
        min_size=5, max_size=20, command_timeout=3
    )
    http_session = aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=2),
        connector=aiohttp.TCPConnector(limit=100, ttl_dns_cache=300)
    )
    yield
    await http_session.close()
    await pool.close()

app.router.lifespan_context = lifespan

async def fetch_logistics(order_id: int) -> dict:
    url = f"http://logistics.internal/api/status?order_id={order_id}"
    try:
        async with http_session.get(url) as resp:
            return await resp.json() if resp.status == 200 else {}
    except (asyncio.TimeoutError, aiohttp.ClientError):
        return {}   # 物流失败降级，不影响主流程

@app.get("/api/order/detail")
async def order_detail(order_id: int):
    async with pool.acquire() as conn:
        # 主表先查，不存在直接返回，避免无谓并发
        row = await conn.fetchrow(
            "SELECT id, user_id, amount, status FROM orders WHERE id=$1", order_id
        )
        if not row:
            raise HTTPException(status_code=404, detail="not found")

        # 三个附属查询 + 外部HTTP 并发执行
        items_task = conn.fetch(
            "SELECT sku, qty, price FROM order_items WHERE order_id=$1", order_id
        )
        pay_task = conn.fetch(
            "SELECT pay_no, paid_at FROM payments WHERE order_id=$1", order_id
        )
        logistics_task = fetch_logistics(order_id)

        items, payments, logistics = await asyncio.gather(
            items_task, pay_task, logistics_task
        )

    return {
        "code": 0,
        "data": {
            "id": row["id"], "user_id": row["user_id"],
            "amount": float(row["amount"]), "status": row["status"],
            "items": [{"sku": r["sku"], "qty": r["qty"], "price": float(r["price"])} for r in items],
            "payments": [{"pay_no": r["pay_no"], "paid_at": str(r["paid_at"])} for r in payments],
            "logistics": logistics,
        },
    }

async def query_items():
    async with pool.acquire() as c:
        return await c.fetch("SELECT sku, qty, price FROM order_items WHERE order_id=$1", order_id)

async def query_payments():
    async with pool.acquire() as c:
        return await c.fetch("SELECT pay_no, paid_at FROM payments WHERE order_id=$1", order_id)