一、问题背景

项目是一个内部数据服务，用 FastAPI 写的一个 /api/v1/orders/summary 接口，给前端看订单聚合数据。上线初期没什么量，最近接入了两个新业务方，QPS 从个位数涨到 50 左右，问题就暴露了：

  • P99 从 200ms 直接飙到 1.8s
  • Postgres 的 CPU 使用率长期在 85%~95%
  • 偶尔出现 TimeoutError，前端开始报错

第一反应是加机器，但看了一下监控，CPU 是单核跑满，明显是有慢查询或者同步阻塞。加机器只是把问题摊平，不解决根因。决定老老实实做一次 profiling。

二、环境与版本

先把环境交代清楚，不同版本行为差异挺大的：

  • Python 3.11.6
  • FastAPI 0.109.2
  • Uvicorn 0.27.0（workers=4，loop=uvloop）
  • SQLAlchemy 2.0.25 + asyncpg 0.29.0
  • PostgreSQL 15.4
  • Redis 7.2.3（redis-py 5.0.1）
  • py-spy 0.3.14
  • Locust 2.20.0

部署在 4C8G 的容器里，Uvicorn 开 4 个 worker。

三、方案设计

整体思路分三步：

  1. 定位：用 py-spy 对 worker 进程做火焰图采样，确认是 CPU 密集还是 IO 等待；同时打开 SQLAlchemy 的 echo 抓实际 SQL。
  2. 优化查询：重点解决 N+1 和缺失索引，把循环里的查询合并成一次 join 或者批量 IN。
  3. 加缓存：对聚合结果按 (user_id, date_range) 做 Redis 缓存，TTL 60s，热点数据基本命中。

下面按顺序讲。

3.1 py-spy 采样定位

生产环境不好直接上 cProfile（开销大且会阻塞），py-spy 是最省事的选择，直接 attach 到进程：

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

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

火焰图打开后一眼就看出来了：sqlalchemy/orm/loading.py 和 asyncpg/protocol 占了大概 70% 的宽度，而且调用栈里反复出现 _load_for_path。典型的 N+1。

顺手也开了 SQL 日志确认：

# 临时打开，压测时观察
engine = create_async_engine(DATABASE_URL, echo=True)

日志里同一接口刷出了 300 多条 SELECT * FROM order_items WHERE order_id = $1。问题坐实。

3.2 原代码的问题

原来的查询大概长这样（简化版）：

@router.get("/orders/summary")
async def order_summary(user_id: int, start: date, end: date, db: AsyncSession = Depends(get_db)):
    stmt = select(Order).where(
        Order.user_id == user_id,
        Order.created_at.between(start, end),
    )
    orders = (await db.execute(stmt)).scalars().all()

    result = []
    for order in orders:
        # 这里触发 N+1
        items = (await db.execute(
            select(OrderItem).where(OrderItem.order_id == order.id)
        )).scalars().all()
        result.append({
            "order_id": order.id,
            "amount": order.amount,
            "item_count": len(items),
        })
    return result

两个问题：

  1. 每个 order 单独查一次 items，N+1。
  2. created_at 上没有索引，between 走全表扫描。

3.3 查询优化

第一步：加索引。

CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);

CONCURRENTLY 是为了不锁表，线上加的。加完之后 EXPLAIN ANALYZE 从 Seq Scan 变成了 Index Scan，单次查询从 240ms 降到 8ms。

第二步：干掉 N+1。

思路是用 selectinload 或者直接 join 聚合。这里其实不需要 item 的明细，只要 count，所以直接一条 SQL 搞定：

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

@router.get("/orders/summary")
async def order_summary(
    user_id: int,
    start: date,
    end: date,
    db: AsyncSession = Depends(get_db),
):
    stmt = (
        select(
            Order.id,
            Order.amount,
            func.count(OrderItem.id).label("item_count"),
        )
        .outerjoin(OrderItem, OrderItem.order_id == Order.id)
        .where(
            Order.user_id == user_id,
            Order.created_at.between(start, end),
        )
        .group_by(Order.id, Order.amount)
        .order_by(Order.created_at.desc())
    )
    rows = (await db.execute(stmt)).all()
    return [
        {"order_id": r.id, "amount": r.amount, "item_count": r.item_count}
        for r in rows
    ]

一条 SQL 出结果，300 次查询变 1 次。这一步之后 P99 从 1.8s 降到大概 320ms。

3.4 缓存策略

300ms 还是不够看，因为这个接口的读远多于写，而且同一用户短时间内会反复刷新。上 Redis 缓存。

缓存 key 设计：order_summary:{user_id}:{start}:{end}，TTL 60 秒。TTL 选 60s 是因为业务上能接受 1 分钟的数据延迟，再短命中率就掉下来了。

import json
from redis.asyncio import Redis

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

async def get_summary_cached(user_id: int, start: date, end: date, db: AsyncSession):
    key = f"order_summary:{user_id}:{start}:{end}"
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)

    data = await _query_summary(user_id, start, end, db)
    # 加个随机抖动，防止同一时刻大量 key 同时过期
    ttl = 60 + random.randint(0, 10)
    await redis.set(key, json.dumps(data, default=str), ex=ttl)
    return data

踩坑一：缓存击穿。

压测时发现某个热点用户过期瞬间，几百个请求同时打到数据库。解决方案是加一个轻量级的分布式锁（SET NX EX），只让一个请求回源，其它等一小会儿再读缓存：

lock_key = f"lock:{key}"
got_lock = await redis.set(lock_key, "1", nx=True, ex=5)
if not got_lock:
    await asyncio.sleep(0.05)
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)
    # 兜底还是走一次 DB

踩坑二：连接池。

原来用的是默认配置，pool_size=5、max_overflow=10，QPS 一上来就报 QueuePool limit of size 5 overflow 10 reached。改成：

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=30,
    pool_pre_ping=True,
    pool_recycle=1800,
)

pool_pre_ping 是为了避免拿到被 PG 服务端关掉的僵尸连接。

踩坑三：Redis 反序列化。

一开始用 json.dumps 默认的 default=str，Date 类型被转成字符串再读出来还是字符串，前端就炸了。后来统一在响应模型里用 Pydantic 做类型转换，缓存里只存原始 dict。

3.5 压测对比

用 Locust 做压测，脚本很简单：

from locust import HttpUser, task, between
import random

class ApiUser(HttpUser):
    wait_time = between(0.01, 0.05)

    @task
    def summary(self):
        uid = random.randint(1, 500)
        self.client.get(
            f"/api/v1/orders/summary?user_id={uid}"
            f"&start=2024-01-01&end=2024-03-31"
        )

4 个 worker，400 并发，跑 3 分钟，结果对比：

阶段 QPS P50 P95 P99 PG CPU
优化前 52 620ms 1.4s 1.8s 92%
加索引后 130 210ms 480ms 620ms 55%
消除 N+1 260 90ms 210ms 320ms 32%
加缓存 420 28ms 75ms 120ms 12%

缓存命中率稳定在 93% 左右。P99 从 1.8s 降到 120ms，差不多 15 倍。

四、踩坑与优化小结

几个值得记下来的点：

  1. 别急着加机器。CPU 单核打满多半是代码问题，先 profiler 再谈扩容。
  2. py-spy 比 cProfile 好用，线上直接 attach，开销可以忽略。
  3. ORM 的 N+1 是重灾区。能用一条 SQL 就别在 Python 里循环查。
  4. 索引不是加得越多越好。(user_id, created_at) 这个复合索引的顺序很关键，因为查询里 user_id 是等值，created_at 是范围，等值在前、范围在后。
  5. 缓存要防击穿。TTL 加随机抖动 + 分布式锁，成本很低但效果显著。
  6. 连接池参数要跟着并发调。默认值在生产环境基本不够用。

五、总结

这次调优从 profiling 到落地大概花了两天，核心动作只有三个：加索引、消 N+1、上缓存。每个动作都有明确的性能收益，没有做任何"猜测式优化"。

如果你的 FastAPI 接口也遇到 P99 偏高的问题，建议按这个顺序来：先 py-spy 定位 → 再看 SQL 日志 → 最后考虑缓存。缓存是最容易见效但也是最容易掩盖问题的，别一上来就加缓存，不然数据库的慢查询会一直藏着，等到缓存大面积失效的时候才发现就晚了。