FastAPI接口P99从1.8s降到120ms：一次真实性能瓶颈定位与优化记录

# 采样 30 秒，输出火焰图
py-spy record -o profile.svg --pid $(pgrep -f "uvicorn app.main:app") --duration 30 --rate 200

import cProfile
import pstats
import io
from fastapi import Request

@app.middleware("http")
async def profile_slow_requests(request: Request, call_next):
    profiler = cProfile.Profile()
    profiler.enable()
    response = await call_next(request)
    profiler.disable()
    if response.headers.get("x-elapsed-ms") and int(response.headers["x-elapsed-ms"]) > 500:
        s = io.StringIO()
        pstats.Stats(profiler, stream=s).sort_stats("cumulative").print_stats(20)
        logger.warning("slow request profile:\n%s", s.getvalue())
    return response

import asyncio
import httpx
from fastapi import APIRouter, HTTPException

router = APIRouter()

# 全局复用连接池，不要每次请求新建 client
_http_client = httpx.AsyncClient(
    timeout=httpx.Timeout(connect=0.5, read=1.5, write=1.5, pool=0.5),
    limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
    http2=True,
)

async def fetch_shipping_template(shop_id: int):
    r = await _http_client.get(f"http://shipping-svc/internal/templates/{shop_id}")
    r.raise_for_status()
    return r.json()

async def fetch_recommendations(product_id: int):
    r = await _http_client.get(f"http://rec-svc/internal/recs/{product_id}")
    r.raise_for_status()
    return r.json()

@router.get("/api/v1/products/{product_id}")
async def get_product(product_id: int):
    # 商品主信息、远程调用、SKU 批量查询三者互不依赖，一起并发
    product_task = asyncio.create_task(load_product_detail(product_id))
    shipping_task = asyncio.create_task(fetch_shipping_template(product_id))
    rec_task = asyncio.create_task(fetch_recommendations(product_id))
    product, shipping, recs = await asyncio.gather(
        product_task, shipping_task, rec_task, return_exceptions=False
    )
    return {"product": product, "shipping": shipping, "recommendations": recs}

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

async def load_product_detail(session: AsyncSession, product_id: int):
    stmt = (
        select(Product)
        .options(selectinload(Product.skus))          # 一条 SQL 批量加载所有 SKU
        .where(Product.id == product_id)
    )
    product = (await session.execute(stmt)).scalar_one_or_none()
    if product is None:
        raise HTTPException(status_code=404, detail="product not found")

    sku_ids = [sku.id for sku in product.skus]
    # 库存和价格一次 IN 查询搞定，替代原来的 for 循环
    stock_rows = await session.execute(
        select(SkuStock.sku_id, SkuStock.available).where(SkuStock.sku_id.in_(sku_ids))
    )
    stock_map = {row.sku_id: row.available for row in stock_rows}
    for sku in product.skus:
        sku.available = stock_map.get(sku.id, 0)
    return product

CREATE INDEX CONCURRENTLY idx_product_sku_product_id ON product_sku (product_id);
CREATE INDEX CONCURRENTLY idx_sku_stock_sku_id ON sku_stock (sku_id);

import json
import random
from cachetools import TTLCache

_local_cache = TTLCache(maxsize=5000, ttl=30)

async def get_product_cached(redis, session, product_id: int):
    key = f"product:detail:{product_id}"
    if (hit := _local_cache.get(key)) is not None:
        return hit

    if raw := await redis.get(key):
        data = json.loads(raw)
        _local_cache[key] = data
        return data

    data = await load_product_detail(session, product_id)
    payload = serialize_product(data)
    # TTL 加 0~60s 抖动，避免同一批 key 同时过期造成缓存击穿
    await redis.set(key, json.dumps(payload), ex=300 + random.randint(0, 60))
    _local_cache[key] = payload
    return payload

async def safe_fetch(coro, default):
    try:
        return await coro
    except (httpx.HTTPError, asyncio.TimeoutError) as e:
        logger.warning("remote call failed: %s", e)
        return default