FastAPI接口P99从1.8s降到120ms：一次profiling驱动的数据库与缓存优化记录

# 找到 uvicorn worker 进程
ps -ef | grep uvicorn

# 对一个 worker 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --rate 200

import time
import logging
from fastapi import Request
from sqlalchemy import event
from sqlalchemy.engine import Engine

logger = logging.getLogger("api.perf")

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start", []).append(time.perf_counter())

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    start = conn.info["query_start"].pop()
    elapsed = (time.perf_counter() - start) * 1000
    if elapsed > 50:  # 只记录超过50ms的
        logger.warning("SLOW SQL %.1fms: %s", elapsed, statement[:200])

async def perf_middleware(request: Request, call_next):
    t0 = time.perf_counter()
    response = await call_next(request)
    cost = (time.perf_counter() - t0) * 1000
    response.headers["X-Resp-Time"] = f"{cost:.1f}ms"
    if cost > 500:
        logger.warning("%s %s cost %.1fms", request.method, request.url.path, cost)
    return response

wrk -t8 -c200 -d60s --latency \
  "http://127.0.0.1:8000/api/v1/products/123456"

# 优化前：每个关联对象都触发一次查询
stmt = select(Product).where(Product.id == product_id)
product = (await session.execute(stmt)).scalar_one()
# 访问 product.skus 时又查一次，product.images 再查一次...

from sqlalchemy.orm import selectinload
from sqlalchemy import select

async def get_product_detail(session, product_id: int):
    stmt = (
        select(Product)
        .where(Product.id == product_id)
        .options(
            selectinload(Product.skus).selectinload(Sku.images),
            selectinload(Product.inventory),
        )
    )
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

import json
import asyncio
import hashlib
from functools import wraps
from cachetools import TTLCache
import redis.asyncio as aioredis

local_cache = TTLCache(maxsize=2048, ttl=5)
redis_client = aioredis.from_url("redis://127.0.0.1:6379/0", decode_responses=True)
_locks: dict[str, asyncio.Lock] = {}

def cached(prefix: str, ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            raw_key = f"{prefix}:{args}:{sorted(kwargs.items())}"
            key = hashlib.md5(raw_key.encode()).hexdigest()

            # L1 本地缓存
            if key in local_cache:
                return local_cache[key]

            # L2 Redis
            cached_val = await redis_client.get(f"{prefix}:{key}")
            if cached_val is not None:
                data = json.loads(cached_val)
                local_cache[key] = data
                return data

            # 防击穿：同key只放一个请求回源
            lock = _locks.setdefault(key, asyncio.Lock())
            async with lock:
                cached_val = await redis_client.get(f"{prefix}:{key}")
                if cached_val is not None:
                    data = json.loads(cached_val)
                    local_cache[key] = data
                    return data

                data = await func(*args, **kwargs)
                # 空值也缓存，防穿透（TTL短一点）
                payload = json.dumps(data, default=str)
                await redis_client.setex(
                    f"{prefix}:{key}",
                    30 if data is None else ttl,
                    payload,
                )
                local_cache[key] = data
                return data
        return wrapper
    return decorator

@app.get("/api/v1/products/{product_id}")
async def product_detail(product_id: int):
    data = await _load_product(product_id)
    if data is None:
        raise HTTPException(status_code=404, detail="product not found")
    return data

@cached(prefix="product:detail", ttl=300)
async def _load_product(product_id: int):
    async with async_session() as session:
        product = await get_product_detail(session, product_id)
        if product is None:
            return None
        # 关键：在会话内完成序列化，避免懒加载在会话外触发
        return ProductSchema.model_validate(product).model_dump()