FastAPI接口响应从1.8s降到86ms：一次Profiling驱动的数据库与缓存优化记录

Type     Name              # reqs   # fails   Median  p95    p99    req/s
GET      /api/v1/products  58213    0.31%    210ms   890ms  1800ms 118.6

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

# 采样30秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --rate 100

# ❌ 改造前：典型N+1
@app.get("/api/v1/products/{product_id}")
async def get_product(product_id: int, db: AsyncSession = Depends(get_db)):
    product = await db.get(Product, product_id)
    skus = await db.execute(select(SKU).where(SKU.product_id == product_id))
    sku_list = skus.scalars().all()
    for sku in sku_list:                      # 每个SKU查一次库存
        sku.stock = await db.scalar(
            select(Stock.quantity).where(Stock.sku_id == sku.id)
        )
    shop = await db.get(Shop, product.shop_id)  # 单独查店铺
    reviews = await db.execute(                   # 单独查评价
        select(Review).where(Review.product_id == product_id)
        .order_by(Review.created_at.desc()).limit(10)
    )
    return build_response(product, sku_list, shop, reviews.scalars().all())

# ✅ 改造后：预加载 + 复合索引 + 只取需要的列
from sqlalchemy.orm import selectinload
from sqlalchemy import select

@app.get("/api/v1/products/{product_id}")
async def get_product(product_id: int, db: AsyncSession = Depends(get_db)):
    stmt = (
        select(Product)
        .options(
            selectinload(Product.skus).selectinload(SKU.stock),
            selectinload(Product.shop),
            selectinload(Product.reviews.and_(
                Review.is_deleted == False
            )),
        )
        .where(Product.id == product_id, Product.is_deleted == False)
    )
    product = (await db.execute(stmt)).scalar_one_or_none()
    if product is None:
        raise HTTPException(status_code=404, detail="product not found")

    # 评价只取最近10条，避免把1200万行的表整个拉进内存
    reviews = sorted(product.reviews, key=lambda r: r.created_at, reverse=True)[:10]
    return build_response(product, reviews)

-- SKU按商品查
CREATE INDEX CONCURRENTLY idx_sku_product_id ON sku (product_id) WHERE is_deleted = false;
-- 库存按SKU查（唯一索引，顺便保证一致性）
CREATE UNIQUE INDEX CONCURRENTLY idx_stock_sku_id ON stock (sku_id);
-- 评价按商品+时间倒序（覆盖排序，避免filesort）
CREATE INDEX CONCURRENTLY idx_review_product_time
    ON review (product_id, created_at DESC) WHERE is_deleted = false;

import json
from cachetools import TTLCache
from redis.asyncio import Redis

# L1：进程内，只存店铺这种小对象，1000条 × 60s
_shop_cache: TTLCache = TTLCache(maxsize=1000, ttl=60)

async def get_shop_cached(shop_id: int, db, redis: Redis) -> dict:
    if (cached := _shop_cache.get(shop_id)) is not None:
        return cached
    key = f"shop:base:{shop_id}:v1"
    if (raw := await redis.get(key)) is not None:
        data = json.loads(raw)
        _shop_cache[shop_id] = data
        return data
    shop = await db.get(Shop, shop_id)
    data = {"id": shop.id, "name": shop.name, "logo": shop.logo}
    await redis.setex(key, 300, json.dumps(data))
    _shop_cache[shop_id] = data
    return data

import asyncio
from collections import defaultdict

_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)

async def get_product_detail_cached(product_id: int, db, redis: Redis) -> dict:
    key = f"product:detail:{product_id}:v3"
    if (raw := await redis.get(key)) is not None:
        return json.loads(raw)

    # 同一进程内同key串行，避免击穿
    async with _locks[key]:
        if (raw := await redis.get(key)) is not None:  # double check
            return json.loads(raw)
        data = await build_product_detail(product_id, db)  # 真实查询
        await redis.setex(key, 60, json.dumps(data, ensure_ascii=False))
        return data