一、问题背景：一个“看起来没毛病”的接口

上周五线上告警，/api/v1/orders/summary 接口P95延迟飙到812ms，QPS只有40，数据库连接数却打到120上限。这个接口是给运营后台的月度报表用的，参数是 month 和 region，逻辑很简单：查订单表、聚合金额、返回JSON。

第一反应是SQL慢，EXPLAIN ANALYZE跑了一遍，主查询只用了38ms。那问题在哪？用 py-spy dump 抓线程栈，发现大量时间耗在 sqlalchemy.orm.loader 和 pydantic.validator 上。典型的N+1查询，一次请求触发了87条SQL，每条2-5ms，加上Python对象序列化，延迟就叠上去了。

环境信息：FastAPI 0.104.1，SQLAlchemy 2.0.23，PostgreSQL 15.3，Redis 7.2，Python 3.11。服务器是4核8G的容器，部署在K8s里，limit配置的是 requests: 500m, limits: 1c。

二、Profiling：先定位，别瞎优化

优化前必须搞清楚时间花在哪了。我用三个工具：

  1. cProfile（离线压测时用）
import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient

client = TestClient(app)
profiler = cProfile.Profile()
profiler.enable()
# 模拟真实请求参数
resp = client.get("/api/v1/orders/summary?month=2024-03&region=华东")
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(40)

输出关键行：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
87     0.002    0.000    0.412    0.005  sqlalchemy/orm/loading.py:82(load_from_identity_map)
87     0.301    0.003    0.301    0.003  sqlalchemy/orm/loading.py:610(_load_scalar_from_collection)
1      0.002    0.002    0.198    0.198  app/schemas/order_schema.py:24(validator)
  1. py-spy（线上时用，不用改代码）
# 找到PID
kubectl exec -it  -- py-spy dump --pid 1
  1. EXPLAIN ANALYZE 验证SQL假设（确认主查询没问题）

结论：主SQL 38ms，但ORM触发了87次额外查询（懒加载关联表），加上Pydantic的validator对每个订单做正则匹配（校验订单号格式），整体耗时被放大。数据库连接被长事务占住，连接池不够用。

三、方案设计：三层优化

第一层：查询优化（干掉N+1）
  - 使用 joinedload 或 selectinload 预加载关联
  - 用 with_entities 只取需要的列，而不是整行ORM对象

第二层：缓存策略（减少重复计算）
  - Redis缓存聚合结果，key设计为 month:region:page
  - 缓存过期时间设为300秒，后台任务主动失效

第三层：序列化优化（减少CPU开销）
  - 绕过Pydantic的validator（用原始dict返回）
  - 或者用 msgspec 替换Pydantic（性能是2倍）

第一层是必须的，第二层看场景，第三层是锦上添花。先做第一层，看效果再决定要不要继续。

四、核心实现：代码改动

4.1 干掉N+1：selectinload + 精简列

原代码（问题代码）：

# app/routers/orders.py
from sqlalchemy.orm import Session
from app.models import Order, OrderItem

def get_order_summary(db: Session, month: str, region: str):
    orders = db.query(Order).filter(
        Order.month == month,
        Order.region == region
    ).all()
    # 这里触发N+1：每次访问 order.items 都会查一次数据库
    return [{
        "order_id": o.id,
        "total": sum(item.price * item.qty for item in o.items),
        "customer": o.customer.name  # 另一张表，又触发查询
    } for o in orders]

优化后：

# app/routers/orders.py
from sqlalchemy.orm import Session, selectinload
from sqlalchemy import func, select
from app.models import Order, OrderItem, Customer

def get_order_summary(db: Session, month: str, region: str):
    # 1. 用selectinload预加载，避免懒加载
    # 2. 只select需要的列，不加载整行对象
    stmt = (
        select(
            Order.id,
            Order.order_no,
            Customer.name.label("customer_name"),
            func.sum(OrderItem.price * OrderItem.qty).label("total")
        )
        .join(Customer, Order.customer_id == Customer.id)
        .join(OrderItem, Order.id == OrderItem.order_id)
        .where(Order.month == month, Order.region == region)
        .group_by(Order.id, Order.order_no, Customer.name)
    )
    results = db.execute(stmt).all()

    return [{
        "order_id": r.id,
        "order_no": r.order_no,
        "customer": r.customer_name,
        "total": round(r.total, 2)
    } for r in results]

改动点：
- 用 select 替代 query，明确指定需要的列
- 用 join + group_by 一次性把关联数据查出来
- 返回的是普通dict，绕过了Pydantic的validator开销

4.2 Redis缓存：写入与失效

# app/services/cache.py
import json
import redis.asyncio as aioredis
from typing import Optional

redis_client = aioredis.from_url(
    "redis://localhost:6379/0",
    max_connections=20,
    socket_timeout=2,
    decode_responses=True
)

CACHE_TTL = 300  # 5分钟

async def get_cached_summary(month: str, region: str) -> Optional[dict]:
    key = f"summary:{month}:{region}"
    data = await redis_client.get(key)
    return json.loads(data) if data else None

async def set_cached_summary(month: str, region: str, data: dict):
    key = f"summary:{month}:{region}"
    await redis_client.setex(key, CACHE_TTL, json.dumps(data))

路由里这样用：

@app.get("/api/v1/orders/summary")
async def summary(month: str, region: str, db: Session = Depends(get_db)):
    # 先查缓存
    cached = await get_cached_summary(month, region)
    if cached:
        return cached
    # 查数据库（同步函数用run_in_executor避免阻塞事件循环）
    data = await asyncio.to_thread(get_order_summary, db, month, region)
    await set_cached_summary(month, region, data)
    return data

五、踩坑与优化记录

坑1：Redis连接池爆炸
一开始用 redis.Redis(connection_pool=...) 每个请求新建连接，压测时直接报 ConnectionError。改成 redis.asyncio 的全局单例后解决。注意 max_connections=20 要小于数据库连接池上限（我们设了35）。

坑2：缓存穿透
月度报表查不到数据时，会直接查数据库，导致缓存没意义。加了空值缓存：

if not data:
    await redis_client.setex(key, 60, json.dumps([]))  # 空数据缓存60秒

坑3：Pydantic validator是隐藏杀手
我们有个全局schema校验订单号格式，用正则 ^[A-Z]{2}\d{8}$。在压测时发现这个正则执行了87次/请求，每次0.1ms，总共8.7ms。优化：只在创建订单时校验，查询时不再走schema。

坑4：数据库连接池参数
SQLAlchemy默认 pool_size=5, max_overflow=10，不够用。调到 pool_size=20, max_overflow=15，但注意容器内存限制。同时加了 pool_pre_ping=True 避免连接失效。

六、压测数据对比

用 wrk 压测，参数：4线程，200连接，60秒，month=2024-03&region=华东。

指标 优化前 优化后（查询+缓存） 优化后（+序列化）
P95延迟 812ms 156ms 96ms
P99延迟 1204ms 289ms 168ms
QPS 42 180 218
数据库连接数 120（打满） 35（稳定） 28
SQL执行次数/请求 87 1 1

最终效果：
- P95从812ms降到96ms，下降88%
- QPS从42提升到218，提升5.2倍
- 数据库连接峰值从120降到28，降低77%

额外收益：Redis缓存命中率在压测场景下约60%（因为压测参数固定），实际生产环境运营只查最近3个月数据，缓存命中率预计90%+。

七、总结与反思

  1. 先profiling再动手，别凭感觉优化SQL。这次问题根本不在SQL，而在ORM懒加载和序列化。
  2. ORM要慎用，复杂聚合查询直接写SQL或 select() 明确列，别怕“不够ORM”。
  3. 缓存是最后手段，如果查询本身优化到1条SQL，其实不需要缓存。但考虑到运营会重复查同一月份数据，缓存还是值得的。
  4. Pydantic慢在validator，如果接口是内部服务，可以跳过schema直接返回dict；如果对外，用msgspec替代。
  5. 监控要跟上，加了缓存后要监控命中率和内存占用，Redis别当垃圾桶什么都塞。

现在这个接口P95稳定在100ms左右，数据库连接压力小了很多。下次碰到类似问题，我会先 py-spy dump 看线程栈，再 EXPLAIN ANALYZE 验证SQL，最后才考虑缓存。