用asyncio重构Flask接口：QPS从120提升到2100的异步化改造记录

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

app = Flask(__name__)

BASE = "http://internal-gateway/api"

@app.route("/api/v1/product/")
def get_product(pid):
    # 三次串行 HTTP 调用，累计耗时 = 80 + 120 + 60 = 260ms 起步
    r1 = requests.get(f"{BASE}/product/{pid}", timeout=1)
    r2 = requests.get(f"{BASE}/stock/{pid}", timeout=1)
    r3 = requests.get(f"{BASE}/price/{pid}", timeout=1)

    if not (r1.ok and r2.ok and r3.ok):
        return jsonify({"error": "upstream failed"}), 502

    return jsonify({
        "product": r1.json(),
        "stock": r2.json(),
        "price": r3.json(),
    })

if __name__ == "__main__":
    # 同步 WSGI 部署，gunicorn -w 4 -k sync 起步
    app.run(port=8000)

# clients.py
import httpx

BASE = "http://internal-gateway/api"

# 全局单例，在 FastAPI lifespan 里创建和关闭
_http_client: httpx.AsyncClient | None = None

def get_client() -> httpx.AsyncClient:
    global _http_client
    if _http_client is None:
        _http_client = httpx.AsyncClient(
            base_url=BASE,
            timeout=httpx.Timeout(0.8, connect=0.3),  # 单请求 800ms，建连 300ms
            limits=httpx.Limits(
                max_connections=300,          # 连接池总上限
                max_keepalive_connections=100, # 长连接保留数
                keepalive_expiry=30.0,
            ),
        )
    return _http_client

async def close_client() -> None:
    global _http_client
    if _http_client is not None:
        await _http_client.aclose()
        _http_client = None

# service.py
import asyncio
import logging
from clients import get_client

logger = logging.getLogger(__name__)

# 按下游分别限流，库存服务最脆，给最小配额
SEM_PRODUCT = asyncio.Semaphore(200)
SEM_STOCK = asyncio.Semaphore(100)
SEM_PRICE = asyncio.Semaphore(200)

PER_CALL_TIMEOUT = 0.8   # 单个下游超时
TOTAL_TIMEOUT = 1.0      # 整体聚合超时

async def _fetch(path: str, sem: asyncio.Semaphore, name: str):
    client = get_client()
    async with sem:
        try:
            resp = await asyncio.wait_for(
                client.get(path), timeout=PER_CALL_TIMEOUT
            )
            resp.raise_for_status()
            return resp.json()
        except (asyncio.TimeoutError, httpx.HTTPError) as e:
            # 降级：记录并返回 None，不拖垮整个请求
            logger.warning("upstream %s failed: %r", name, e)
            return None

async def get_product_detail(pid: int) -> dict:
    # 三个下游并发发出
    tasks = [
        _fetch(f"/product/{pid}", SEM_PRODUCT, "product"),
        _fetch(f"/stock/{pid}", SEM_STOCK, "stock"),
        _fetch(f"/price/{pid}", SEM_PRICE, "price"),
    ]
    try:
        product, stock, price = await asyncio.wait_for(
            asyncio.gather(*tasks), timeout=TOTAL_TIMEOUT
        )
    except asyncio.TimeoutError:
        logger.error("aggregate timeout for pid=%s", pid)
        return {"error": "aggregate timeout"}

    # 商品信息是必需的，缺了就算失败；库存/价格可降级
    if product is None:
        return {"error": "product unavailable"}

    return {
        "product": product,
        "stock": stock,
        "price": price,
    }

# app_async.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from clients import close_client
from service import get_product_detail

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await close_client()

app = FastAPI(lifespan=lifespan)

@app.get("/api/v1/product/{pid}")
async def product(pid: int):
    return await get_product_detail(pid)

gunicorn app_async:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  -b 0.0.0.0:8000 \
  --timeout 30