一、问题背景

先说结论：一个看起来只有几十行逻辑的 GET /products/{id} 接口，在压测下 P95 干到了 1.8 秒，QPS 只有 230 左右就开始大面积 504。这个接口本身逻辑很简单——查商品详情、查分类、查几个 SKU、再拼一下库存。但就是这种"看起来简单"的接口最容易埋雷。

我们当时的业务场景是：大促前压测，商品详情页是流量最大的入口，占比大概 60%。测试同学反馈"接口慢",开发第一反应是"加机器"，但我在看监控时发现 CPU 只跑到 40%，DB 的 CPU 却接近 90%。这就说明问题不在应用层算力，而在数据访问路径。

所以这篇文章讲的不是"如何做一个高性能 API"这种空话，而是：怎么用 profiling 找到真正的瓶颈、怎么判断是 N+1 还是缓存缺失、怎么改、改完数据是多少。

二、环境与版本

先把环境交代清楚，不然数据没法复现：

  • Python 3.11.6
  • FastAPI 0.110.0
  • Uvicorn 0.29.0（--workers 4）
  • SQLAlchemy 2.0.29（async 模式，asyncpg 0.29.0）
  • PostgreSQL 15.6
  • Redis 7.2.4（redis-py 5.0.3）
  • 压测工具：wrk 4.2.0 + locust 2.24.0（做阶梯压测）
  • 机器：4C8G 容器，DB 单独一台 8C16G

接口原始实现大概是这样（简化版）：

# app/api/product.py (优化前)
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.db import get_session
from app.models import Product, Category, SKU

router = APIRouter()

@router.get("/products/{pid}")
async def get_product(pid: int, db: AsyncSession = Depends(get_session)):
    product = (await db.execute(
        select(Product).where(Product.id == pid)
    )).scalar_one_or_none()
    if not product:
        return {"code": 404, "msg": "not found"}

    category = (await db.execute(
        select(Category).where(Category.id == product.category_id)
    )).scalar_one()

    skus = (await db.execute(
        select(SKU).where(SKU.product_id == pid)
    )).scalars().all()

    sku_list = []
    for s in skus:
        sku_list.append({
            "id": s.id,
            "name": s.name,
            "price": float(s.price),
            "stock": s.stock,
        })

    return {
        "code": 0,
        "data": {
            "id": product.id,
            "title": product.title,
            "category": category.name,
            "skus": sku_list,
        }
    }

这段代码一眼看过去没什么毛病，select 写得也挺规范。但它在压测下的问题会在后面暴露出来。

三、方案设计：先定位，再动手

我给自己定了个顺序，避免"凭感觉优化"：

  1. 先用 py-spy + cProfile 定位热点函数，确认是 CPU 密集还是 IO 等待。
  2. 打开 SQLAlchemy 的 echo 或慢查询日志，看一次请求到底发了几条 SQL。
  3. 确认是 N+1、缺索引，还是缺缓存。
  4. 按"成本从低到高"改：先加索引 → 再改查询 → 再上缓存 → 最后调连接池。

3.1 用 py-spy 抓火焰图

py-spy 的好处是不用改代码、不用重启，直接 attach 到运行中的进程：

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

# 或者直接 top 看实时热点
py-spy top --pid 12345

第一次看火焰图就很明显：大量时间卡在 asyncpg 的 _wait_for_response 上。也就是说，应用层没在算，全在等数据库。这就直接排除了"Python 代码写得慢"这个方向。

3.2 用 cProfile 看函数级耗时

对单个请求做精确统计，我用 cProfile 包了一层：

# scripts/prof_one_request.py
import asyncio
import cProfile
import pstats
import io
from app.api.product import get_product
from app.db import AsyncSessionLocal

async def run():
    async with AsyncSessionLocal() as db:
        await get_product(1, db)

def main():
    pr = cProfile.Profile()
    pr.enable()
    asyncio.run(run())
    pr.disable()
    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
    ps.print_stats(20)
    print(s.getvalue())

if __name__ == "__main__":
    main()

输出里 asyncpg 相关调用累计占了 80% 以上，确认瓶颈在 DB。

3.3 看 SQL 数量

把 SQLAlchemy 日志级别调到 INFO，或者临时加 echo=True，一次请求打出来 4 条 SQL：

SELECT ... FROM products WHERE id = 1;
SELECT ... FROM categories WHERE id = 3;
SELECT ... FROM skus WHERE product_id = 1;
SELECT ... FROM skus WHERE product_id = 1;   -- 又查了一次

前三条还算正常，第四条是 ORM 关系懒加载触发的。这就是典型的 N+1 的变种：因为 product.skus 在别处被访问了一次，SQLAlchemy 又发了一次查询。

四、核心实现：三步优化

4.1 第一步：消除重复查询 + 加索引

用 selectinload 一次性把关联对象拉出来，同时给 skus.product_id 加索引：

# app/api/product.py (优化后)
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from app.db import get_session
from app.models import Product
from app.cache import redis_client
import json

router = APIRouter()

@router.get("/products/{pid}")
async def get_product(pid: int, db: AsyncSession = Depends(get_session)):
    cache_key = f"product:detail:{pid}"
    cached = await redis_client.get(cache_key)
    if cached:
        return {"code": 0, "data": json.loads(cached), "cached": True}

    stmt = (
        select(Product)
        .options(selectinload(Product.category), selectinload(Product.skus))
        .where(Product.id == pid)
    )
    product = (await db.execute(stmt)).scalar_one_or_none()
    if not product:
        raise HTTPException(status_code=404, detail="not found")

    data = {
        "id": product.id,
        "title": product.title,
        "category": product.category.name,
        "skus": [
            {
                "id": s.id,
                "name": s.name,
                "price": float(s.price),
                "stock": s.stock,
            }
            for s in product.skus
        ],
    }

    # 缓存 300 秒，带随机抖动防雪崩
    import random
    ttl = 300 + random.randint(0, 30)
    await redis_client.setex(cache_key, ttl, json.dumps(data, ensure_ascii=False))
    return {"code": 0, "data": data, "cached": False}

索引：

CREATE INDEX CONCURRENTLY idx_skus_product_id ON skus (product_id);
CREATE INDEX CONCURRENTLY idx_products_category_id ON products (category_id);

到这里，单请求 SQL 从 4 条降到 2 条（selectinload 会合并成两条：一条主表 + 一条 in 查询）。

4.2 第二步：Redis 缓存

上面代码里已经加了缓存。几个关键点：

  • TTL 加随机抖动：防止大量 key 同时过期造成缓存雪崩。
  • 缓存穿透：对不存在的 pid，可以缓存空值 __NULL__ 短 TTL，避免每次都打 DB。
  • 序列化：用 json.dumps 而不是 pickle，跨语言安全，也避免反序列化漏洞。

Redis 连接池配置（这个很关键，默认配置在并发高时会成为瓶颈）：

# app/cache.py
import redis.asyncio as redis

redis_client = redis.Redis(
    host="127.0.0.1",
    port=6379,
    db=0,
    max_connections=200,        # 默认是 2**31，但实际 socket 数会爆
    socket_timeout=0.5,         # 超时快速失败，避免拖垮请求
    socket_connect_timeout=0.5,
    retry_on_timeout=True,
    health_check_interval=30,
    decode_responses=True,
)

4.3 第三步：DB 连接池与 Uvicorn 参数

SQLAlchemy async 引擎配置：

# app/db.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pwd@db:5432/shop",
    pool_size=20,            # 每个 worker 的连接池大小
    max_overflow=10,
    pool_pre_ping=True,      # 防止拿到死连接
    pool_recycle=1800,       # 30 分钟回收，配合 DB 的 idle timeout
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

Uvicorn 启动：

uvicorn app.main:app \
  --host 0.0.0.0 --port 8000 \
  --workers 4 \
  --loop uvloop \
  --http httptools \
  --backlog 2048 \
  --limit-concurrency 1000

pool_size * workers 要小于 DB 的 max_connections，否则高并发下会报 too many clients。我们 DB 设的是 max_connections=200，4 个 worker × 20 = 80，留足余量。

五、踩坑与优化

坑 1：缓存和 DB 数据不一致。
一开始我们在写操作里直接 del cache，但并发下会出现"删缓存 → 读旧值回填"。后来改成先更新 DB 再删缓存，并且写操作加一个短锁（Redis SETNX，TTL 3 秒），保证同一商品的写串行。

坑 2：selectinload 拉太多数据。
selectinload(Product.skus) 会把该商品所有 SKU 都拉出来。如果某个商品有几百个 SKU，单次响应体就会很大。后来加了 .limit(50) 和分页参数。

坑 3：连接池耗尽导致请求堆积。
压测初期 P99 突然从 200ms 涨到 5s，看日志发现是 QueuePool limit of size 20 overflow 10 reached。原因是缓存命中率低的时候所有请求都打到 DB。解决办法是：提高缓存命中率 + 增加连接池 + 设置 socket_timeout 快速失败，不要让请求无限等待。

坑 4：py-spy 在容器里 attach 需要 --cap-add=SYS_PTRACE。
这个不踩一次真的会懵，报 Permission denied。K8s 里可以加：

securityContext:
  capabilities:
    add: ["SYS_PTRACE"]

六、效果数据

压测条件：4C8G 容器 × 1，wrk 并发 200，持续 3 分钟，数据集 10 万商品，每个商品平均 8 个 SKU。

指标 优化前 优化后
P50 620 ms 45 ms
P95 1800 ms 120 ms
P99 3100 ms 210 ms
QPS 230 1420
单请求 SQL 数 4 0（命中缓存）/ 2（未命中）
缓存命中率 - 96.3%
DB CPU 90% 25%
错误率 4.7% 0%

locust 阶梯压测下，优化后 QPS 到 1400 时 P95 依然稳定在 130ms 以内，DB 连接池使用率最高 65%。

说句实话，这个提升里 缓存贡献了大概 70%，selectinload 和索引贡献了 20%，连接池和 uvicorn 参数贡献 10%。所以如果只让我做一件事，我会先看缓存命中率。

七、总结

这次调优的核心不是某个"银弹"，而是一个可复用的排查链路：

  1. py-spy 火焰图 → 判断是 CPU 还是 IO 瓶颈；
  2. SQL 日志 / cProfile → 确认是 N+1 还是慢查询；
  3. 索引 + selectinload → 消除重复查询；
  4. Redis 缓存 → 挡住绝大部分读流量；
  5. 连接池 + uvicorn 参数 → 保证高并发下不雪崩。

如果你的接口也慢，别急着加机器，先跑一遍 py-spy。很多时候你会发现：不是机器不够，是 SQL 发太多了。

最后留个 checklist，方便你对着排查：

  • [ ] 单请求 SQL 数量是否 > 3？
  • [ ] 有没有 N+1（懒加载）？
  • [ ] 关联字段有没有索引？
  • [ ] 热点数据有没有缓存？TTL 有没有抖动？
  • [ ] 缓存穿透、雪崩、击穿是否处理？
  • [ ] DB 连接池和 worker 数是否匹配？
  • [ ] 有没有慢查询日志？P95 是多少？

以上数据都是我们线上真实压测的结果，环境和参数都写在文里了，有兴趣的同学可以照着复现一遍。有更好的方案欢迎评论区交流。