一、问题背景：生产环境告警

那是个周五晚上，监控面板突然跳出红色告警：/api/v1/orders 接口P95延迟从正常的180ms飙升至2340ms，错误率1.2%，数据库连接数打满。查看慢查询日志，发现一条关联了5张表的SQL执行了1.8秒，而这条SQL在ORM层被触发了187次（N+1问题）。更离谱的是，这个接口的QPS只有40，意味着每次请求触发了将近5次数据库往返。

我们当时的架构是：Nginx → Uvicorn(4 workers) → FastAPI(0.95.0) → PostgreSQL 15，ORM使用SQLAlchemy 2.0.19，缓存层是Redis 7.0。老板给的压力很大：“明天早上客户要演示，必须修复。”

二、环境与版本

调优前的基础环境：

  • Python 3.11.4
  • FastAPI 0.95.0 + Uvicorn 0.23.2 (workers=4, loop=uvloop)
  • SQLAlchemy 2.0.19 + asyncpg 0.27.0
  • PostgreSQL 15.3 (shared_buffers=4GB, effective_cache_size=12GB)
  • Redis 7.0.11 (maxmemory=2GB, maxmemory-policy=allkeys-lru)
  • 压测工具：wrk 4.2.0 + Locust 2.15.1
  • Profiling工具：PySpy 0.3.14 + cProfile + pg_stat_statements

三、第一步：Profiling定位瓶颈

3.1 用PySpy抓取在线调用栈

Uvicorn的worker进程是异步的，直接attach PySpy可能看不到Python层调用。先安装：

pip install pyspy
# 找到worker进程PID
ps aux | grep uvicorn
# attach到CPU使用率最高的那个worker
sudo pyspy dump --pid 12345 --duration 30

输出片段（关键部分）：

Thread 0x7f8d2c1b6700 (idle):
  - task: Task was created at:
    - FastAPI route /api/v1/orders (order_router.py:42)
    - ...
  - Current frame:
    - asyncpg.protocol.protocol: _process_message (asyncpg/protocol/protocol.pyx:590)
    - sqlalchemy.dialects.postgresql.asyncpg: _executemany (dialects/postgresql/asyncpg.py:580)
    - sqlalchemy.orm.session: flush (session.py:2800)
    - app.services.order_service: get_orders (order_service.py:88)

看到_executemany和flush，立刻怀疑是ORM的懒加载导致批量写入或查询。再用cProfile做一次离线复现验证。

3.2 cProfile离线分析

写一个复现脚本，模拟真实请求逻辑：

import cProfile, pstats, asyncio
from app.services.order_service import get_orders

async def main():
    # 模拟真实参数
    await get_orders(user_id=1024, page=1, page_size=20)

if __name__ == "__main__":
    profiler = cProfile.Profile()
    profiler.enable()
    asyncio.run(main())
    profiler.disable()
    stats = pstats.Stats(profiler).sort_stats('cumulative')
    stats.print_stats(30)

输出统计（截取前几行）：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
    187    0.012    0.000    1.723    0.009  sqlalchemy/orm/loading.py:180(load_on_ident)
    187    1.412    0.008    1.412    0.008  asyncpg/protocol/protocol.pyx:590(_process_message)
      1    0.003    0.003    1.731    1.731  order_service.py:88(get_orders)
      1    0.001    0.001    0.521    0.521  jsonable_encoder.py:45(jsonable_encoder)

实锤：187次load_on_ident（懒加载），累计1.723秒，占整个请求的80%。还有0.5秒在JSON序列化。两个瓶颈：N+1查询 + 序列化开销。

四、方案设计：三管齐下

定位到两个瓶颈后，设计如下优化方案：

  1. 消除N+1：SQLAlchemy 2.0的selectinload批量加载关联表，把187次查询压到3次（主查询 + 2个关联查询）。
  2. 增加Redis缓存：对热点数据（用户最近订单）做缓存，TTL=60秒，key设计为orders:{user_id}:{page}:{page_size}，使用JSON序列化后存储。
  3. 优化序列化：用msgspec替代Pydantic的jsonable_encoder（实测快3.2倍）。

五、核心实现：代码修改

5.1 SQLAlchemy查询重构（关键代码）

重构前的代码（有问题的版本）：

# order_service.py (优化前)
async def get_orders(user_id: int, page: int, page_size: int):
    async with async_session() as session:
        # 这里只查了orders表，items和products是懒加载
        result = await session.execute(
            select(Order).where(Order.user_id == user_id)
            .offset((page-1)*page_size).limit(page_size)
        )
        orders = result.scalars().all()
        # 访问order.items时触发N+1查询
        return [
            {
                "id": o.id,
                "items": [
                    {"id": item.id, "product": item.product.name}
                    for item in o.items  # 每次访问触发一次查询
                ]
            } for o in orders
        ]

重构后使用selectinload + joinedload组合：

# order_service.py (优化后)
from sqlalchemy.orm import selectinload, joinedload
from sqlalchemy import select

SELECTIN_OPTIONS = (
    selectinload(Order.items).selectinload(OrderItem.product),
)

async def get_orders(user_id: int, page: int, page_size: int):
    async with async_session() as session:
        # 一次性加载所有关联，查询次数 = 1(主) + 1(items) + 1(products) = 3次
        result = await session.execute(
            select(Order)
            .options(*SELECTIN_OPTIONS)
            .where(Order.user_id == user_id)
            .order_by(Order.created_at.desc())
            .offset((page-1)*page_size)
            .limit(page_size)
        )
        orders = result.scalars().unique().all()
        return [
            {
                "id": o.id,
                "created_at": o.created_at.isoformat(),
                "items": [
                    {"id": item.id, "product_name": item.product.name}
                    for item in o.items  # 此时已经是内存数据
                ]
            } for o in orders
        ]

5.2 Redis缓存层（防御性缓存）

# cache.py
import msgspec
import redis.asyncio as aioredis
from fastapi import HTTPException

# 使用msgspec的JSON解码器，比json.loads快3倍
json_decoder = msgspec.json.Decoder(type=list[dict])
json_encoder = msgspec.json.Encoder()

redis_client = aioredis.from_url(
    "redis://localhost:6379/0",
    max_connections=50,
    decode_responses=False  # 保持bytes模式，msgspec直接处理
)

CACHE_TTL = 60  # 秒
CACHE_PREFIX = "orders:v2"

async def get_cached_orders(user_id: int, page: int, page_size: int):
    cache_key = f"{CACHE_PREFIX}:{user_id}:{page}:{page_size}"
    cached = await redis_client.get(cache_key)
    if cached:
        return json_decoder.decode(cached)
    return None

async def set_cached_orders(user_id: int, page: int, page_size: int, data: list[dict]):
    cache_key = f"{CACHE_PREFIX}:{user_id}:{page}:{page_size}"
    encoded = json_encoder.encode(data)
    await redis_client.setex(cache_key, CACHE_TTL, encoded)

在FastAPI路由中接入缓存（注意处理缓存穿透）：

# order_router.py
from fastapi import APIRouter, Query, Depends
from .cache import get_cached_orders, set_cached_orders

router = APIRouter(prefix="/api/v1")

@router.get("/orders")
async def get_orders_endpoint(
    user_id: int = Query(..., gt=0),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100)
):
    # 先查缓存
    cached = await get_cached_orders(user_id, page, page_size)
    if cached is not None:
        return {"data": cached, "source": "cache"}

    # 缓存未命中，查数据库
    data = await get_orders(user_id, page, page_size)
    if not data:
        # 防止缓存穿透：空结果也缓存10秒
        await set_cached_orders(user_id, page, page_size, data, ttl=10)
        return {"data": data, "source": "db"}

    await set_cached_orders(user_id, page, page_size, data)
    return {"data": data, "source": "db"}

5.3 Flask版本对比（同样逻辑）

为了对比，我也用Flask 3.0写了一个等价接口。Flask本身是同步WSGI，用gunicorn+gevent跑。查询逻辑用同样的SQLAlchemy 2.0（同步版），缓存逻辑一样。核心区别：FastAPI走asyncpg原生异步驱动，Flask走psycopg2同步驱动。

六、踩坑与优化记录

坑1：selectinload 导致笛卡尔积

一开始我用了joinedload加载items和product，结果返回行数爆炸（20个订单×每个订单5个item×每个item1个product = 100行），数据量翻倍。后来改成selectinload，它先查主表，再WHERE id IN (...)查关联表，避免笛卡尔积。

坑2：缓存击穿

上线后第一波流量直接打崩Redis——所有用户同时过期，全部穿透到数据库。加了随机过期时间修复：

import random
# 在setex时加抖动
await redis_client.setex(cache_key, CACHE_TTL + random.randint(0, 15), encoded)

坑3：Pydantic v2的序列化陷阱

FastAPI默认用jsonable_encoder序列化ORM对象，它会把datetime转成字符串，但这个过程很慢。改用msgspec后，直接在ORM层把数据整理成dict再编码，省去了中间转换。

七、压测数据对比

使用wrk压测，参数：wrk -t4 -c100 -d30s --latency http://localhost:8000/api/v1/orders?user_id=1024&page=1&page_size=20

配置 P50 (ms) P95 (ms) P99 (ms) QPS 错误率
优化前(FastAPI+懒加载) 892 2314 3890 41 0.8%
优化后(FastAPI+selectinload) 305 782 1201 156 0.0%
优化后+Redis缓存 87 287 512 237 0.0%
Flask(同步+selectinload) 420 1103 1870 98 0.2%
Flask+Redis缓存 205 612 940 132 0.1%

结论：FastAPI + selectinload + Redis 组合拳，P95从2314ms降到287ms（降幅87.6%），QPS从41涨到237（提升5.8倍）。Flask方案虽然也有优化，但受限于同步驱动，P95始终比FastAPI慢2倍左右。

补充一个数据点：使用pg_stat_statements查看，优化后单接口的SQL执行次数从187次/请求降到3次/请求，数据库CPU使用率从95%降到22%。

八、总结与心得

这次调优的核心收获：

  1. 先测量再动手：用PySpy+cProfile 5分钟定位瓶颈，比瞎猜快100倍。不要一上来就加索引、换框架。
  2. ORM懒加载是性能杀手：任何ORM框架（SQLAlchemy、Django ORM、Hibernate）都有N+1问题，必须显式使用selectinload或joinedload。
  3. 缓存是银弹但不是万能的：Redis缓存解决读多写少场景，但要注意穿透、击穿、雪崩。实测缓存命中率91%时，效果最好。
  4. FastAPI vs Flask：如果追求极致性能且团队熟悉异步，选FastAPI。如果是内部管理系统、QPS<100，Flask完全够用且调试更简单。
  5. 序列化开销被低估：如果响应体超过100KB，序列化可能占CPU的20%。用msgspec或orjson替代默认编码器。

最后，把压测脚本和监控告警都接入了CI/CD，以后每次改动API都会自动跑性能回归，再也没出现过半夜被叫醒的情况。

参考链接：
- SQLAlchemy 2.0 relationship loading: https://docs.sqlalchemy.org/en/20/orm/loading_relationships.html
- PySpy GitHub: https://github.com/benfred/py-spy
- msgspec文档: https://jcristharif.com/msgspec/
- FastAPI性能优化官方文档: https://fastapi.tiangolo.com/advanced/performance/