FastAPI+SQLAlchemy 2.0 性能调优：从1200ms到80ms的profiling与缓存实践

1. 问题背景：一个“看起来没问题”的接口

事情发生在上个月，业务方反馈某个订单列表接口在高峰期频繁超时（>1s），并且数据库CPU经常飙到90%。我接手排查时，第一反应是加索引，但查看慢查询日志后发现：这个接口压根没有慢SQL，而是发了300+条小查询。

典型的ORM反模式——N+1查询。因为业务用的是FastAPI + SQLAlchemy 2.0，我当时用了relationship的懒加载，导致每查一条订单就要额外查询一次关联的商家和用户表。

另外，响应体里有大量冗余字段（比如完整的用户对象、商家对象），JSON序列化耗时也占了总耗时的20%左右。

先给出环境版本，方便你复现：

Python 3.11.5
FastAPI 0.104.1
SQLAlchemy 2.0.23
Redis 7.2（用redis-py 5.0）
wrk 4.2.0（压测工具）

2. Profiling：别猜，用工具说话

我一开始凭直觉认为瓶颈在数据库索引，但优化后效果甚微。后来用py-spy直接attach到线上进程，抓了30秒的火焰图：

# 安装
pip install py-spy

# 抓取线上进程（PID 12345）的火焰图，输出SVG
py-spy record --pid 12345 --output profile.svg --duration 30

火焰图结果非常直观：sqlalchemy.orm.query.Query.iterate_instances 占了61%的CPU,pydantic.BaseModel.__init__占了19%。这说明两个问题：

ORM懒加载导致大量重复查询（61% CPU）
Pydantic序列化开销过大（19% CPU）

为了进一步验证N+1，我在本地用cProfile跑了一个最小复现：

import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient

client = TestClient(app)

def call_api():
    for _ in range(100):
        client.get("/api/v1/orders?page=1&size=20")

profiler = cProfile.Profile()
profiler.enable()
call_api()
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(30)

输出中明显看到sqlalchemy.orm.strategies.LazyLoader._emit_lazy_load被调用了超过3000次，平均每次1.2ms，合计约3.6秒——这就是1200ms响应时间的核心来源。

3. 方案设计：三级降级策略

基于profiling结果，我设计了如下优化顺序（由易到难）：

第一级：ORM查询优化（消除N+1）——用selectinload或joinedload预加载关联对象，预计耗时降低60%。
第二级：序列化优化——改用orm_mode + 自定义响应模型，只返回必要字段，预计降低15%。
第三级：Redis缓存（Cache-Aside模式）——对热点订单页做缓存，TTL 5分钟，预计降低剩余耗时的80%。

为什么不用lru_cache？因为接口需要根据page和size动态生成，且数据在商家端偶尔会更新，lru_cache无法主动失效，最后一致性很难保证。Redis + 逻辑过期（TTL内主动刷新）更可控。

4. 核心实现：三步走

第一步：SQLAlchemy 2.0查询优化

原代码（懒惰加载）：

# 原代码：懒加载
async def get_orders(db: Session, page: int, size: int):
    orders = db.execute(
        select(Order).order_by(Order.created_at.desc())
        .offset((page-1)*size).limit(size)
    ).scalars().all()
    return orders

优化后（selectinload预加载）：

# 优化后：selectinload一次性加载关联对象
from sqlalchemy.orm import selectinload

async def get_orders(db: Session, page: int, size: int):
    stmt = (
        select(Order)
        .options(
            selectinload(Order.user),   # 预加载用户
            selectinload(Order.merchant) # 预加载商家
        )
        .order_by(Order.created_at.desc())
        .offset((page-1) * size)
        .limit(size)
    )
    orders = db.execute(stmt).scalars().all()
    return orders

注意：selectinload会生成第二条WHERE id IN (...)查询，配合limit使用效率远高于joinedload（因为joinedload会把结果集膨胀）。

第二步：Pydantic响应模型瘦身

定义只包含必要字段的响应模型：

from pydantic import BaseModel, ConfigDict

class UserBrief(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    nickname: str
    avatar: str

class MerchantBrief(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str

class OrderOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    order_no: str
    amount: float
    status: int
    user: UserBrief
    merchant: MerchantBrief

然后在路由中声明response_model=List[OrderOut]，FastAPI会自动过滤掉多余字段。这一步看似简单，但能减少约30%的序列化时间。

第三步：Redis Cache-Aside缓存

import json
from redis import asyncio as aioredis

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

# 缓存key设计：orders:{page}:{size}
async def get_orders_cached(db: Session, page: int, size: int):
    cache_key = f"orders:{page}:{size}"

    # 先查缓存
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)  # 返回反序列化后的列表

    # 缓存未命中，查数据库
    orders = await get_orders(db, page, size)

    # 转换为可JSON序列化的dict列表
    order_dicts = [OrderOut.model_validate(o).model_dump(mode="json") for o in orders]

    # 写入缓存，TTL 300秒
    await redis_client.setex(cache_key, 300, json.dumps(order_dicts))
    return order_dicts

注意：这里用了model_dump(mode="json")而不是json()，因为后者返回的是字符串，还需要二次解析。mode="json"直接产出原生Python类型，配合json.dumps效率更高。

5. 踩坑与优化：三个意想不到的坑

坑1：selectinload与分页的“假命中”

我一开始用joinedload配合limit，结果发现返回的数据只有10条（预期20条）。原因是joinedload会产生笛卡尔积，导致LIMIT作用于错误的结果集。换成selectinload后问题消失，因为SQLAlchemy 2.0会拆分成两条SQL，第一条只查主表分页，第二条用WHERE id IN (上页的id列表)查关联表。

坑2：Pydantic v2的from_attributes兼容性

升级到Pydantic v2后，原先的orm_mode=True变成了ConfigDict(from_attributes=True)。如果不改，会直接报AttributeError。建议直接升级到v2，因为v2的序列化速度比v1快3-5倍。

坑3：Redis缓存穿透

压测时发现，当请求page=999时，缓存永远为空，每次都打到数据库。加了一个简单的空值缓存：

if cached is None:
    # 查数据库后，如果结果为空，缓存空列表，TTL短一点（60秒）
    if not order_dicts:
        await redis_client.setex(cache_key, 60, json.dumps([]))

6. 效果数据：前后对比

用wrk压测（单机8线程200连接，持续30秒）：

# 优化前
wrk -t8 -c200 -d30s http://localhost:8000/api/v1/orders?page=1&size=20
Latency: avg 1180ms, max 2200ms, P95 1450ms
QPS: 52

# 优化后（查询优化+序列化瘦身）
Latency: avg 320ms, max 600ms, P95 410ms
QPS: 210

# 最终（+Redis缓存）
Latency: avg 78ms, max 150ms, P95 95ms
QPS: 820

数据库侧的表现：

# 优化前
单接口平均SQL查询次数：302次
数据库CPU：85-95%

# 优化后
单接口平均SQL查询次数：2次（selectinload拆成2条）
数据库CPU：12-15%

注意，缓存命中率约75%（因为TTL只有5分钟，且订单数据变动频繁）。如果业务上能接受10分钟TTL，QPS还能再翻一倍。

7. 总结与建议

三条核心经验：

永远先profiling再优化。我用py-spy只花了10分钟就找到N+1，而之前靠猜浪费了半天。
FastAPI + SQLAlchemy 2.0的selectinload是分页场景的首选，别用joinedload。
缓存是最后的手段，不要一开始就上。先消除N+1和序列化瓶颈，缓存效果才能最大化。

如果你用的是Flask + SQLAlchemy，思路完全一样，只是把async def换成普通函数，Redis客户端换成同步的redis.Redis。核心在于：减少查询次数 + 减少传输数据量 + 缓存热数据。

最后补一句：如果你的接口响应体特别大（>100KB），建议考虑gzip压缩，FastAPI自带GZipMiddleware，直接加一行代码的事。我这个接口响应体约20KB，压缩后传输时间降了30%。

大家在看

FastAPI+SQLAlchemy 2.0 性能调优：从1200ms到80ms的profiling与缓存实践

1. 问题背景：一个“看起来没问题”的接口

2. Profiling：别猜，用工具说话

3. 方案设计：三级降级策略

4. 核心实现：三步走

5. 踩坑与优化：三个意想不到的坑

6. 效果数据：前后对比

7. 总结与建议

评论 1

📖 相关推荐

Git分支策略与CI/CD流水线：团队协作的版本控制实践

7B模型LoRA微调全流程：数据、训练、Loss与推理对比

FastAPI异步重构与Redis缓存：API响应从2.3秒降至95ms

FastAPI性能调优：Profiling、SQL查询与缓存策略压测实录