用asyncio重构Web API：QPS从120提升到1800的优化记录

# before_api.py
import requests
import psycopg2
from fastapi import FastAPI

app = FastAPI()

DB_CONN = psycopg2.connect(
    host="10.0.1.20", port=5432, dbname="shop",
    user="api", password="xxx"
)

@app.get("/order/{user_id}")
def get_order_detail(user_id: int):
    # 串行调用三个下游
    order_resp = requests.get(f"http://order-svc/orders?user_id={user_id}", timeout=3)
    orders = order_resp.json()

    user_resp = requests.get(f"http://user-svc/users/{user_id}", timeout=3)
    user = user_resp.json()

    # 同步DB查询
    cur = DB_CONN.cursor()
    cur.execute("SELECT sku_id, stock FROM inventory WHERE user_id = %s", (user_id,))
    stocks = cur.fetchall()
    cur.close()

    return {"user": user, "orders": orders, "stocks": stocks}

# after_api.py
import asyncio
import asyncpg
import httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException

# 全局资源
http_client: httpx.AsyncClient = None
pg_pool: asyncpg.Pool = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client, pg_pool
    # 连接池参数：根据下游承载能力调，max_connections别超过下游限流
    limits = httpx.Limits(
        max_connections=200,
        max_keepalive_connections=50,
        keepalive_expiry=30.0,
    )
    timeout = httpx.Timeout(connect=0.5, read=2.0, write=1.0, pool=1.0)
    http_client = httpx.AsyncClient(limits=limits, timeout=timeout, http2=True)

    pg_pool = await asyncpg.create_pool(
        host="10.0.1.20", port=5432, database="shop",
        user="api", password="xxx",
        min_size=10, max_size=50,
        command_timeout=2.0,
    )
    yield
    await http_client.aclose()
    await pg_pool.close()


app = FastAPI(lifespan=lifespan)


async def fetch_orders(user_id: int):
    resp = await http_client.get(f"http://order-svc/orders", params={"user_id": user_id})
    resp.raise_for_status()
    return resp.json()


async def fetch_user(user_id: int):
    resp = await http_client.get(f"http://user-svc/users/{user_id}")
    resp.raise_for_status()
    return resp.json()


async def fetch_stocks(user_id: int):
    async with pg_pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT sku_id, stock FROM inventory WHERE user_id = $1", user_id
        )
    return [dict(r) for r in rows]


@app.get("/order/{user_id}")
async def get_order_detail(user_id: int):
    # 并发执行，return_exceptions=True 保证单个失败不影响整体
    results = await asyncio.gather(
        fetch_orders(user_id),
        fetch_user(user_id),
        fetch_stocks(user_id),
        return_exceptions=True,
    )

    orders, user, stocks = results
    degraded = []

    if isinstance(orders, Exception):
        degraded.append("orders")
        orders = []
    if isinstance(user, Exception):
        degraded.append("user")
        user = {"user_id": user_id, "name": "unknown"}
    if isinstance(stocks, Exception):
        degraded.append("stocks")
        stocks = []

    # 全挂才算失败；部分失败返回降级数据 + 标记
    if len(degraded) == 3:
        raise HTTPException(status_code=503, detail="all upstreams unavailable")

    return {
        "user": user,
        "orders": orders,
        "stocks": stocks,
        "degraded": degraded,
    }