FastAPI+Flask接口从1.2s降到86ms：一次SQLAlchemy查询与Redis缓存的性能调优记录

一、问题背景：一个看起来很简单的接口，为什么这么慢

事情是这样的。我们有个订单聚合接口 /api/v1/orders/summary，前端在用户中心页面加载时调用，返回用户近30天的订单统计、最近5条订单详情、以及每单的商品数量。逻辑不复杂，但上线后监控上一直很难看：

• P95 响应时间：1.2s（Sentry 上报）
• P99：2.4s
• 用 Locust 压测，80 QPS 时错误率就开始飙升，主要是超时
• 数据库 CPU 在高峰期打到 70%+

接口本身是 FastAPI 写的，但项目里还有几个老接口是 Flask 的，共用同一套 SQLAlchemy 模型和 MySQL 实例。所以这次调优其实是两个框架的公共问题：ORM 层和缓存层。

我一开始的猜测是数据库慢查询，但具体是哪条、慢在哪，得靠数据说话，不能拍脑袋。

二、环境与版本

先把环境列清楚，免得后面数据对不上：

• Python 3.11.6
• FastAPI 0.110.0 + Uvicorn 0.27.1（4 workers，gunicorn 管理）
• Flask 3.0.2（老接口，WSGI 走 gunicorn sync worker）
• SQLAlchemy 2.0.28（ORM 模式，非 2.0 style 全量迁移）
• MySQL 8.0.35，InnoDB，buffer pool 4G
• Redis 7.2.4，单机，maxmemory 2G，allkeys-lru
• Locust 2.24.0 做压测
• py-spy 0.3.14 做采样 profiling

服务器：8C16G，接口服务和 MySQL 不在同一台机器上，内网 RTT 约 0.3ms。

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

我的思路分三步走，不跳步：

1. Profiling 定位：先用 py-spy 抓火焰图看 CPU 时间花在哪，再打开 SQLAlchemy 的 echo 看实际 SQL 条数。两个信息一对照，瓶颈基本就藏不住了。
2. 数据库层优化：针对慢查询做索引、改写 N+1、必要时用 selectinload 预加载。
3. 缓存层优化：对读多写少、容忍短时不一致的数据上 Redis，设计好 key 和 TTL，避免缓存穿透和雪崩。

压测方案：用 Locust 模拟 200 并发用户，持续 3 分钟，对比优化前后的 P50/P95/P99、QPS、错误率。

四、核心实现

4.1 Profiling：py-spy + SQLAlchemy echo

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

# 找到 uvicorn worker 的 pid
ps -ef | grep uvicorn

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

火焰图一出来就很明显：60% 以上的采样落在 sqlalchemy/orm/loading.py 和 pymysql 的 read 上，说明时间几乎全花在等数据库返回。

接着在测试环境打开 echo，把实际 SQL 打出来：

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    "mysql+pymysql://user:pwd@10.0.0.12:3306/order_db?charset=utf8mb4",
    pool_size=20,
    max_overflow=10,
    pool_recycle=3600,
    echo=True,  # 临时打开，观察 SQL
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

请求一次接口，日志里刷出来 47 条 SQL。其中：

• 1 条查用户
• 1 条查订单列表（30 天）
• 30 条查每单的商品（典型 N+1）
• 1 条聚合统计
• 剩下十几条是权限、配置之类的零碎查询

N+1 是主犯，聚合统计那条 SQL 也没走索引，是帮凶。

4.2 查询优化：消灭 N+1 + 加索引

先看原来的代码（简化版）：

# 优化前
@app.get("/api/v1/orders/summary")
def orders_summary(user_id: int, db: Session = Depends(get_db)):
    orders = db.query(Order).filter(
        Order.user_id == user_id,
        Order.created_at >= datetime.now() - timedelta(days=30),
    ).all()

    result = []
    for o in orders:
        # 每次循环都打一次数据库，N+1 元凶
        item_count = db.query(func.count(OrderItem.id)).filter(
            OrderItem.order_id == o.id
        ).scalar()
        result.append({"order_id": o.id, "amount": o.amount, "item_count": item_count})
    return {"orders": result}

改写思路：用 selectinload 一次性把关联的 items 拉出来，再在 Python 里聚合。SQLAlchemy 2.0 里我更推荐 selectinload 而不是 joinedload，因为一对多 join 会导致结果集膨胀，selectin 是发一条 IN 查询，干净。

# 优化后
from sqlalchemy.orm import selectinload

@app.get("/api/v1/orders/summary")
def orders_summary(user_id: int, db: Session = Depends(get_db)):
    stmt = (
        db.query(Order)
        .options(selectinload(Order.items))  # 一条 IN 查询搞定所有 items
        .filter(
            Order.user_id == user_id,
            Order.created_at >= datetime.now() - timedelta(days=30),
        )
        .order_by(Order.created_at.desc())
        .all()
    )

    result = [
        {
            "order_id": o.id,
            "amount": o.amount,
            "item_count": len(o.items),  # 内存里算，不再打库
        }
        for o in stmt
    ]
    return {"orders": result}

SQL 从 47 条直接降到 4 条。

索引这边，orders 表原来的索引只有主键和 user_id 单列索引，但查询条件是 user_id + created_at 范围，单列索引选择性不够。加上联合索引：

ALTER TABLE orders
  ADD INDEX idx_user_created (user_id, created_at DESC);

用 EXPLAIN 验证，type 从 ref 变成 range，rows 从约 12000 降到 32，Extra 里也没了 Using filesort。

4.3 缓存策略：只缓存该缓存的

缓存不是万能药，上错了反而增加不一致和运维成本。我的原则：

• 读多写少：订单统计这类聚合数据，写入频率低，适合缓存
• 容忍短时不一致：统计口径允许 60 秒延迟，用户无感
• 缓存对象小：单个用户的 summary，序列化后 2-5KB

Redis key 设计：

order:summary:{user_id}:{date_bucket}

用日期做 bucket，天然带过期语义。TTL 设 60 秒，加随机抖动 0-10 秒防雪崩。

import json
import random
from redis import Redis

redis_client = Redis(host="10.0.0.20", port=6379, db=0, decode_responses=True)

def get_orders_summary_cached(user_id: int, db: Session):
    bucket = datetime.now().strftime("%Y%m%d%H%M")  # 分钟级 bucket
    key = f"order:summary:{user_id}:{bucket}"

    cached = redis_client.get(key)
    if cached:
        return json.loads(cached)

    data = _query_orders_summary(user_id, db)  # 走上面优化后的查询

    # TTL 60s + 0~10s 抖动，避免同一时刻集体过期
    ttl = 60 + random.randint(0, 10)
    redis_client.setex(key, ttl, json.dumps(data, default=str))
    return data

关于缓存穿透：如果用户没有任何订单，data 是空列表，也会被缓存，这没问题。但如果是恶意构造不存在的 user_id，每次都会穿透到 DB。我在入口加了个轻量布隆过滤器（用 pybloom_live），不过我们用户 ID 是自增的，其实用 user_id <= max_user_id 判断就够了，没必要上布隆，杀鸡用牛刀。

4.4 Flask 老接口同步改造

项目里几个 Flask 老接口有同样的问题，它们是 WSGI 同步模型，不能像 FastAPI 那样 async，但缓存逻辑可以复用：

# Flask 接口复用同一套缓存函数
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route("/api/v1/orders/summary")
def orders_summary_flask():
    user_id = request.args.get("user_id", type=int)
    if not user_id:
        return jsonify({"error": "user_id required"}), 400

    db = SessionLocal()
    try:
        data = get_orders_summary_cached(user_id, db)
        return jsonify(data)
    finally:
        db.close()

Flask 这边 sync worker 的并发能力本来就有限，缓存命中率上去之后，DB 压力下降，效果立竿见影。

五、踩坑与优化

坑 1：expire_on_commit=False 导致的脏读

一开始我在 session 配置里设了 expire_on_commit=False，本意是减少 commit 后的额外查询，但缓存和 ORM 混用时，如果对象在事务里被改过又没 commit，缓存里可能写进旧数据。后来改成缓存层只接收纯 dict，不传 ORM 对象，问题解决。

坑 2：selectinload 的 IN 查询参数过多

有个别用户 30 天内有 500+ 订单，selectinload 会生成 WHERE order_id IN (...) 带 500 个参数，MySQL 的 max_allowed_packet 虽然够，但解析慢。我加了分页限制，summary 接口最多返回 100 条，超出部分前端翻页。

坑 3：Redis 连接池没设上限

最初用 Redis() 默认连接池，压测时发现 Redis 侧连接数飙到 300+，触发 maxclients 告警。改成显式连接池：

pool = redis.ConnectionPool(
    host="10.0.0.20", port=6379, db=0,
    max_connections=50, decode_responses=True,
)
redis_client = Redis(connection_pool=pool)

坑 4：缓存 key 里的日期 bucket 精度

一开始用秒级 bucket，等于没缓存，命中率不到 5%。改成分钟级后，命中率稳定在 92% 左右。

六、效果数据

Locust 压测，200 并发，3 分钟，同一台压测机：

缓存命中率：92.3%（Redis INFO stats 里的 keyspace_hits / (hits+misses)）。

线上灰度一周，P95 稳定在 90ms 上下，DB 慢查询日志从日均 300+ 条降到个位数。

七、总结

这次调优没什么黑科技，全是基本功：

1. 先 profiling 再动手。py-spy 火焰图 + SQLAlchemy echo 两个工具，10 分钟就能把瓶颈锁定，比看代码猜快得多。
2. N+1 是 ORM 性能的头号杀手。selectinload 基本是标配，配合联合索引，SQL 数从 47 降到 4。
3. 缓存要克制。只缓存读多写少、容忍不一致的数据，key 带 bucket、TTL 带抖动，命中率才能上去。
4. FastAPI 和 Flask 共用一套数据层，缓存和查询优化的收益是两个框架一起吃的。

如果你们项目里也有类似接口，建议先跑一遍 py-spy record，说不定瓶颈就在你没想到的地方。