FastAPI+Flask双栈API性能调优：从1200ms到85ms的profiling、SQL与缓存实践

# 找到 gunicorn worker 的 pid
ps -ef | grep gunicorn | grep -v master

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

# profile_api.py
import cProfile, pstats, io
from app import create_app

app = create_app()

def run():
    with app.test_client() as c:
        for _ in range(50):
            c.get("/api/v1/products/10086")

pr = cProfile.Profile()
pr.enable()
run()
pr.disable()

s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
ps.print_stats(25)
print(s.getvalue())

-- 索引（先确认已有主键，这里补的是过滤和排序字段）
CREATE INDEX CONCURRENTLY idx_products_category_status
    ON products (category_id, status) WHERE status = 'on_sale';
CREATE INDEX CONCURRENTLY idx_product_images_pid_sort
    ON product_images (product_id, sort_order);

-- 合并查询
SELECT
    p.id, p.title, p.price, p.status, p.updated_at,
    COALESCE(imgs.images, '[]'::json) AS images,
    COALESCE(skus.skus, '[]'::json)   AS skus,
    COALESCE(attrs.attrs, '[]'::json) AS attrs
FROM products p
LEFT JOIN LATERAL (
    SELECT json_agg(json_build_object('id', id, 'url', url, 'sort', sort_order)
                    ORDER BY sort_order) AS images
    FROM product_images WHERE product_id = p.id
) imgs ON TRUE
LEFT JOIN LATERAL (
    SELECT json_agg(json_build_object('id', id, 'name', name, 'stock', stock)) AS skus
    FROM product_skus WHERE product_id = p.id
) skus ON TRUE
LEFT JOIN LATERAL (
    SELECT json_agg(json_build_object('key', attr_key, 'value', attr_value)) AS attrs
    FROM product_attrs WHERE product_id = p.id
) attrs ON TRUE
WHERE p.id = $1;

# app/api/product.py
import asyncio, json, random
from fastapi import APIRouter, HTTPException
import asyncpg, httpx
from cachetools import TTLCache

router = APIRouter()
local_cache = TTLCache(maxsize=2000, ttl=5)   # 本地热点缓存，5s

SQL = """..."""  # 上面那条合并查询

async def get_pool(app):
    return app.state.pg_pool

async def fetch_recommend(pid: int) -> list:
    """异步调下游推荐服务，带超时和降级"""
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(0.3)) as client:
            r = await client.get(f"http://rec-svc/api/rec/{pid}")
            r.raise_for_status()
            return r.json().get("items", [])
    except Exception:
        return []          # 降级：推荐失效不影响主流程

@router.get("/api/v1/products/{pid}")
async def product_detail(pid: int, request):
    # L1 本地缓存
    if pid in local_cache:
        return local_cache[pid]

    # L2 Redis
    redis = request.app.state.redis
    cached = await redis.get(f"prod:{pid}")
    if cached:
        data = json.loads(cached)
        local_cache[pid] = data
        return data

    pool = await get_pool(request.app)
    async with pool.acquire() as conn:
        row = await conn.fetchrow(SQL, pid)
    if not row:
        raise HTTPException(404, "product not found")

    detail = dict(row)
    # 并行：推荐服务与数据组装同时进行
    rec_task = asyncio.create_task(fetch_recommend(pid))
    detail["recommend"] = await rec_task

    payload = json.dumps(detail, default=str)
    # TTL 加随机抖动，避免同一批 key 同时过期
    await redis.setex(f"prod:{pid}", 60 + random.randint(0, 10), payload)
    local_cache[pid] = detail
    return detail

async def invalidate(pid: int, redis):
    await redis.delete(f"prod:v2:{pid}")
    local_cache.pop(pid, None)

async def init_conn(conn):
    await conn.set_type_codec(
        "json", encoder=json.dumps, decoder=json.loads, schema="pg_catalog"
    )
    await conn.set_type_codec(
        "jsonb", encoder=json.dumps, decoder=json.loads, schema="pg_catalog"
    )

pool = await asyncpg.create_pool(dsn, init=init_conn, min_size=5, max_size=20)