FastAPI+Flask双框架API性能调优：从2.3s到180ms的profiling、SQL与缓存改造

pip install py-spy==0.3.14

# 找到 uvicorn worker 的 pid
ps -ef | grep uvicorn

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

# 或者直接看 top，快速定位热点函数
py-spy top --pid 12345

# 优化前的代码（反面教材）
async def get_product_detail(product_id: int, db: AsyncSession):
    product = await db.get(Product, product_id)
    result = {
        "id": product.id,
        "name": product.name,
        "skus": [],
    }
    # 这里循环查 SKU，一个商品平均 12 个 SKU
    for sku in product.skus:
        stock = await db.execute(
            select(Stock).where(Stock.sku_id == sku.id)
        )
        result["skus"].append({
            "sku_id": sku.id,
            "price": sku.price,
            "stock": stock.scalar_one().quantity,
        })
    return result

from sqlalchemy.orm import selectinload
from sqlalchemy import select

async def get_product_detail(product_id: int, db: AsyncSession):
    stmt = (
        select(Product)
        .options(
            selectinload(Product.skus).selectinload(Sku.stock),
            selectinload(Product.shop),
        )
        .where(Product.id == product_id)
    )
    product = (await db.execute(stmt)).scalar_one_or_none()
    if product is None:
        return None
    return {
        "id": product.id,
        "name": product.name,
        "shop": {"id": product.shop.id, "name": product.shop.name},
        "skus": [
            {
                "sku_id": s.id,
                "price": str(s.price),
                "stock": s.stock.quantity,
            }
            for s in product.skus
        ],
    }

-- 生产环境用 CONCURRENTLY 避免锁表
CREATE INDEX CONCURRENTLY idx_stock_sku_id ON stock (sku_id);
CREATE INDEX CONCURRENTLY idx_sku_product_id ON sku (product_id);

-- 促销标签的联合索引，查询条件是 product_id + status + 时间范围
CREATE INDEX CONCURRENTLY idx_promo_product_status
ON promotion (product_id, status, start_time DESC);

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

# 本地缓存：最多 2000 个 key，TTL 5 秒
_local_cache: TTLCache = TTLCache(maxsize=2000, ttl=5)
_redis: Redis = Redis(host="redis.internal", port=6379, decode_responses=True)

def _key(product_id: int) -> str:
    return f"product:detail:v3:{product_id}"

async def get_product_cached(product_id: int, db) -> dict:
    key = _key(product_id)

    # L1 本地缓存
    if (cached := _local_cache.get(key)) is not None:
        return cached

    # L2 Redis
    if (raw := await _redis.get(key)) is not None:
        data = json.loads(raw)
        _local_cache[key] = data
        return data

    # 回源
    data = await get_product_detail(product_id, db)
    if data is None:
        # 空值缓存，防止穿透，TTL 短一些
        await _redis.setex(key, 30, "null")
        return None

    # 随机 TTL 300~360 秒，防雪崩
    import random
    ttl = 300 + random.randint(0, 60)
    await _redis.setex(key, ttl, json.dumps(data, ensure_ascii=False))
    _local_cache[key] = data
    return data

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,          # 稳态连接数
    max_overflow=10,       # 峰值可临时扩到 30
    pool_pre_ping=True,    # 防连接失效
    pool_recycle=1800,     # 30 分钟回收，避免被 DB 端 kill
    echo=False,
)

wrk -t8 -c200 -d60s --latency http://api.internal/api/v1/products/12345