FastAPI接口P99从1.8s降到180ms：一次profiling驱动的查询与缓存改造

wrk -t8 -c100 -d60s --latency http://127.0.0.1:8000/api/v1/products/feed?category_id=12&page=1&size=20

py-spy record -o profile.svg --pid  --duration 30

from sqlalchemy import event
from sqlalchemy.engine import Engine
import logging, time

logger = logging.getLogger("slow_sql")

@event.listens_for(Engine, "before_cursor_execute")
def before_execute(conn, cursor, statement, params, context, executemany):
    conn.info["query_start"] = time.perf_counter()

@event.listens_for(Engine, "after_cursor_execute")
def after_execute(conn, cursor, statement, params, context, executemany):
    elapsed = (time.perf_counter() - conn.info["query_start"]) * 1000
    if elapsed > 50:
        logger.warning("SLOW %.1fms | %s | params=%s", elapsed, statement[:300], params)

# 优化前：每个商品单独查标签、库存、评价
products = db.query(Product).filter(Product.category_id == cid).offset(offset).limit(size).all()
result = []
for p in products:
    tags = db.query(Tag).join(ProductTag).filter(ProductTag.product_id == p.id).all()
    stock = db.query(Stock).filter(Stock.product_id == p.id).first()
    review = db.query(Review).filter(Review.product_id == p.id).order_by(Review.created_at.desc()).first()
    result.append(build_item(p, tags, stock, review))

from sqlalchemy import select
from sqlalchemy.orm import Session

def get_feed(db: Session, category_id: int, page: int, size: int):
    offset = (page - 1) * size
    # 1. 主查询：只取需要的列
    stmt = (
        select(Product.id, Product.title, Product.price, Product.cover)
        .where(Product.category_id == category_id, Product.status == 1)
        .order_by(Product.id.desc())
        .offset(offset).limit(size)
    )
    rows = db.execute(stmt).all()
    if not rows:
        return []
    pids = [r.id for r in rows]

    # 2. 批量查标签
    tag_rows = db.execute(
        select(ProductTag.product_id, Tag.name)
        .join(Tag, Tag.id == ProductTag.tag_id)
        .where(ProductTag.product_id.in_(pids))
    ).all()
    tag_map = {}
    for pid, name in tag_rows:
        tag_map.setdefault(pid, []).append(name)

    # 3. 批量查库存
    stock_rows = db.execute(
        select(Stock.product_id, Stock.quantity, Stock.status)
        .where(Stock.product_id.in_(pids))
    ).all()
    stock_map = {pid: (q, s) for pid, q, s in stock_rows}

    # 4. 批量查最近评价：用窗口函数一次搞定
    from sqlalchemy import func
    rn = func.row_number().over(
        partition_by=Review.product_id,
        order_by=Review.created_at.desc()
    ).label("rn")
    subq = select(Review.product_id, Review.content, Review.rating, rn).subquery()
    review_rows = db.execute(
        select(subq.c.product_id, subq.c.content, subq.c.rating)
        .where(subq.c.rn == 1, subq.c.product_id.in_(pids))
    ).all()
    review_map = {pid: (c, r) for pid, c, r in review_rows}

    return [build_item(r, tag_map, stock_map, review_map) for r in rows]

-- 主查询覆盖排序和过滤
ALTER TABLE product ADD INDEX idx_cat_status_id (category_id, status, id DESC);

-- 评价窗口函数依赖
ALTER TABLE review ADD INDEX idx_product_created (product_id, created_at DESC);

-- 标签关联
ALTER TABLE product_tag ADD INDEX idx_product (product_id);

import json
from redis import Redis

redis = Redis(host="127.0.0.1", port=6379, decode_responses=True)

def get_feed_cached(db: Session, category_id: int, page: int, size: int):
    # 只缓存前两页，后面页命中率太低不值得
    if page > 2:
        return get_feed(db, category_id, page, size)

    key = f"feed:{category_id}:{page}:{size}:v3"
    cached = redis.get(key)
    if cached:
        return json.loads(cached)

    data = get_feed(db, category_id, page, size)
    redis.setex(key, 60, json.dumps(data, ensure_ascii=False, default=str))
    return data