FastAPI接口P99从1.8s降到120ms：一次 profiling 驱动的查询与缓存重构

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

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

# 优化前：N+1 查询
@app.get("/api/v1/products/{product_id}")
def get_product(product_id: int, db: Session = Depends(get_db)):
    product = db.query(Product).filter(Product.id == product_id).first()
    if not product:
        raise HTTPException(404, "product not found")

    skus = db.query(SKU).filter(SKU.product_id == product_id).all()
    result = []
    for sku in skus:
        # 每个 SKU 单独查库存，N+1 的根源
        stock = db.query(Stock).filter(Stock.sku_id == sku.id).first()
        result.append({
            "sku_id": sku.id,
            "price": str(sku.price),
            "stock": stock.available if stock else 0,
        })
    return {"id": product.id, "name": product.name, "skus": result}

# 优化后：一次 JOIN 查询
from sqlalchemy import select, join
from sqlalchemy.orm import Session

@app.get("/api/v1/products/{product_id}")
def get_product(product_id: int, db: Session = Depends(get_db)):
    stmt = (
        select(
            Product.id, Product.name,
            SKU.id.label("sku_id"), SKU.price,
            Stock.available,
        )
        .join(SKU, SKU.product_id == Product.id)
        .outerjoin(Stock, Stock.sku_id == SKU.id)
        .where(Product.id == product_id)
    )
    rows = db.execute(stmt).all()
    if not rows:
        raise HTTPException(404, "product not found")

    return {
        "id": rows[0].id,
        "name": rows[0].name,
        "skus": [
            {"sku_id": r.sku_id, "price": str(r.price), "stock": r.available or 0}
            for r in rows
        ],
    }

import json
import random
import time
import redis
from fastapi import HTTPException

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
CACHE_TTL = 300
LOCK_TTL = 5

def get_product_cached(product_id: int, db) -> dict:
    key = f"product:detail:{product_id}"
    cached = r.get(key)
    if cached is not None:
        data = json.loads(cached)
        if data == "__NULL__":
            raise HTTPException(404, "product not found")
        return data

    lock_key = f"lock:{key}"
    # SET NX 抢锁，只有抢到的请求回源
    if r.set(lock_key, "1", nx=True, ex=LOCK_TTL):
        try:
            # double check，防止抢锁期间别人已经写好了缓存
            cached = r.get(key)
            if cached is not None:
                return json.loads(cached)

            data = query_product_from_db(product_id, db)  # 上面 4.1 的查询逻辑
            ttl = CACHE_TTL + random.randint(-30, 30)
            if data is None:
                r.set(key, json.dumps("__NULL__"), ex=60)
                raise HTTPException(404, "product not found")
            r.set(key, json.dumps(data), ex=ttl)
            return data
        finally:
            r.delete(lock_key)
    else:
        # 没抢到锁，短暂自旋等待缓存写入
        for _ in range(20):
            time.sleep(0.05)
            cached = r.get(key)
            if cached is not None:
                return json.loads(cached)
        # 兜底：等不到就直接查库，避免无限等待
        return query_product_from_db(product_id, db)

# locustfile.py
import random
from locust import HttpUser, task, between

class ProductUser(HttpUser):
    wait_time = between(0.01, 0.05)

    @task
    def detail(self):
        pid = random.randint(1, 5000)  # 模拟热点分散
        self.client.get(f"/api/v1/products/{pid}", name="/products/{id}")