一、问题背景：报警从早上七点开始

先说场景。我们是一个电商中台团队，服务分两块：

  • 对外网关层：FastAPI 0.110.0 + Uvicorn 0.29.0，Python 3.11.6，负责鉴权、限流、聚合下游服务。
  • 内部订单服务：Flask 3.0.2 + Gunicorn 21.2.0（4 worker × 4 线程），SQLAlchemy 2.0.29，MySQL 8.0.35。

出问题的是订单列表接口 GET /api/v1/orders?user_id=xxx&page=1&size=20。这个接口平时 P99 在 400ms 左右，某天早上七点开始，P99 直接冲到 3.8s，错误率虽然没涨，但超时（>3s）比例到了 12%。

第一反应是流量涨了。看监控：QPS 从 300 涨到 340，涨了 13%，但延迟翻了近 10 倍。这明显不是纯流量问题，是某个环节的耗时被放大了。

我按照老规矩，先看三个东西：

  1. 机器指标：CPU 从 35% 涨到 78%，但没打满；内存正常；MySQL 的 CPU 到 60%，慢查询数从 0 涨到每分钟 40+。
  2. 接口链路：网关层耗时 30ms，订单服务耗时 3.5s，瓶颈在订单服务内部。
  3. MySQL 慢查询日志：SELECT * FROM orders WHERE user_id = ? ORDER BY created_at DESC LIMIT 20 OFFSET 0，扫描行数 80 万，Using filesort。

到这里基本能猜个八九不离十了：user_id 有索引，但排序字段 created_at 没进联合索引，导致 filesort；加上分页偏移和 N+1 查询，数据量一涨就崩。

但我不想拍脑袋改，还是得走一遍 profiling，确认时间到底花在哪。

二、环境与版本

组件 版本
Python 3.11.6
FastAPI 0.110.0
Uvicorn 0.29.0
Flask 3.0.2
Gunicorn 21.2.0
SQLAlchemy 2.0.29
MySQL 8.0.35
Redis 7.2.4
py-spy 0.3.14
locust 2.24.1

压测机是一台 8C16G 的阿里云 ECS，被压的服务是 4C8G，MySQL 是独立的 8C16G 实例。压测时保证没有其他业务干扰，这点很重要，不然数据没法看。

三、方案设计：先测量，再优化

我的思路很朴素，就三步：

  1. 定位：用 py-spy 做采样分析，找出 CPU 热点；用 cProfile 做函数级耗时统计，找出时间黑洞。
  2. 拆解：把一次请求拆成「DB 查询」「业务逻辑」「序列化」三段，分别计时。
  3. 优化：按投入产出比排序，先做收益大、风险低的——索引优化 > 批量查询 > 缓存。

这里有个原则：不要一上来就上缓存。缓存是最后手段，因为一旦上了缓存，数据一致性问题会跟着来，维护成本高。先把 SQL 和代码本身的问题解决掉，再考虑缓存。

四、核心实现：profiling 到优化

4.1 用 py-spy 抓 CPU 热点

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

# 找到 gunicorn worker 的 pid
ps -ef | grep gunicorn | grep -v grep

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

# 或者直接 top 看实时热点
py-spy top --pid

火焰图出来后，最宽的那一坨就是问题所在。我当时看到的是：

  • sqlalchemy/orm/loading.py 占了 42% 的采样
  • json/encoder.py 占了 18%
  • pymysql/connections.py 占了 25%

这说明大量时间花在 ORM 加载和 JSON 序列化上，而 ORM 加载又是因为 N+1 查询——每个订单都要去查一次用户信息。

4.2 用 cProfile 定位函数级耗时

py-spy 只能看 CPU，看不了 IO 等待。我用 cProfile 补上这块，在 Flask 里加了个临时的 profiling 装饰器：

import cProfile
import pstats
import io
from functools import wraps
from flask import request

def profile_route(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 只对特定请求做 profiling，避免污染线上
        if request.args.get("__profile__") != "1":
            return func(*args, **kwargs)
        pr = cProfile.Profile()
        pr.enable()
        result = func(*args, **kwargs)
        pr.disable()
        s = io.StringIO()
        ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
        ps.print_stats(30)
        # 这里实际是打到日志里，线上别直接返回
        print(s.getvalue())
        return result
    return wrapper

跑一次请求后，输出里最扎眼的是：

ncalls  tottime  cumtime  function
    21    0.002    2.841  get_order_detail  # 每个订单查一次详情
     1    0.001    3.102  list_orders

21 次 get_order_detail，每次 135ms 左右，加起来 2.8s。这就是 N+1 查询的铁证。

4.3 优化一：SQL 与索引

先说原始 SQL 的问题：

-- 原始查询
SELECT * FROM orders 
WHERE user_id = 12345 
ORDER BY created_at DESC 
LIMIT 20 OFFSET 0;

user_id 上有单列索引，但 ORDER BY created_at 需要 filesort。改成联合索引：

-- 联合索引，注意顺序：等值条件在前，排序字段在后
ALTER TABLE orders ADD INDEX idx_user_created (user_id, created_at DESC);

-- 如果还要按状态过滤，再加一列
ALTER TABLE orders ADD INDEX idx_user_status_created (user_id, status, created_at DESC);

加完索引后，EXPLAIN 从 type: ref, rows: 800000, Extra: Using filesort 变成 type: ref, rows: 20, Extra: Using index condition。扫描行数从 80 万降到 20，这一步就砍掉了 60% 的耗时。

然后是 N+1 查询。原来的代码：

# 反例：N+1 查询
orders = Order.query.filter_by(user_id=user_id).order_by(Order.created_at.desc()).limit(20).all()
result = []
for order in orders:
    # 每个订单查一次用户，21 次查询
    user = User.query.get(order.user_id)
    result.append({
        "order_id": order.id,
        "user_name": user.name,
        "amount": order.amount,
    })

改成批量查询 + join：

# 优化后：一次 join 拿完
from sqlalchemy.orm import joinedload

orders = (
    db.session.query(Order)
    .options(joinedload(Order.user))  # 预加载关联
    .filter(Order.user_id == user_id)
    .order_by(Order.created_at.desc())
    .limit(20)
    .all()
)

# 如果关联表数据量大，用 in_ 批量查更稳
order_ids = [o.id for o in orders]
details = OrderDetail.query.filter(OrderDetail.order_id.in_(order_ids)).all()
detail_map = {d.order_id: d for d in details}

这里有个坑：joinedload 在数据量大时会产生笛卡尔积，如果一对多关系多，结果集膨胀得厉害。我们订单和订单详情是一对一，所以没问题；如果是一对多，建议用 selectinload，它会先查主表再 WHERE id IN (...)，更可控。

4.4 优化二：Redis 缓存

索引和批量查询做完，P99 降到 620ms。但还有优化空间——这个接口读多写少，热点用户（比如运营账号）会反复查同样的数据，适合加缓存。

我设计了两级缓存：

  • L1：进程内缓存，用 cachetools.TTLCache，容量 10000，TTL 30 秒，扛瞬时热点。
  • L2：Redis，TTL 5 分钟，扛跨进程共享。
import json
import hashlib
from cachetools import TTLCache
from redis import Redis

local_cache = TTLCache(maxsize=10000, ttl=30)
redis_client = Redis(host="redis.internal", port=6379, db=0, socket_timeout=0.05)

def get_cache_key(user_id: int, page: int, size: int) -> str:
    raw = f"orders:{user_id}:{page}:{size}"
    return hashlib.md5(raw.encode()).hexdigest()

def get_orders_cached(user_id: int, page: int, size: int):
    key = get_cache_key(user_id, page, size)

    # L1
    if key in local_cache:
        return local_cache[key]

    # L2
    try:
        cached = redis_client.get(key)
        if cached:
            data = json.loads(cached)
            local_cache[key] = data
            return data
    except Exception:
        # Redis 挂了不能影响主流程，降级到 DB
        pass

    # 回源
    data = query_orders_from_db(user_id, page, size)

    # 回写，注意异常吞掉
    try:
        redis_client.setex(key, 300, json.dumps(data, default=str))
        local_cache[key] = data
    except Exception:
        pass

    return data

这里有几个细节值得说：

  1. socket_timeout 设 50ms，Redis 一旦慢，直接降级，不能拖垮接口。
  2. 缓存穿透：对空结果也缓存，但 TTL 短一点（60 秒），防止恶意刷不存在的 user_id。
  3. 缓存雪崩：TTL 加了 ±10% 的随机抖动，避免同一时刻大面积失效。
  4. 序列化用 default=str，因为订单里有 Decimal 和 datetime，不加会报错。

4.5 FastAPI 侧：异步阻塞问题

网关层虽然只占 30ms，但我也顺手查了一下。发现一个典型问题：FastAPI 的异步路由里调了同步的 requests：

# 反例：async 路由里跑同步阻塞代码
@app.get("/api/v1/orders")
async def get_orders(user_id: int):
    # requests 是同步的，会阻塞事件循环
    resp = requests.get(f"http://order-service/orders?user_id={user_id}", timeout=2)
    return resp.json()

改成 httpx 的异步客户端：

import httpx

client = httpx.AsyncClient(timeout=2.0, limits=httpx.Limits(max_connections=200))

@app.get("/api/v1/orders")
async def get_orders(user_id: int):
    resp = await client.get(f"http://order-service/orders?user_id={user_id}")
    return resp.json()

这个改动看起来小，但在高并发下差别巨大。同步 requests 会把事件循环堵死，Uvicorn 的并发优势完全发挥不出来。改完后网关层 P99 从 30ms 降到 12ms。

五、踩坑与优化

这部分说几个实际踩的坑，比代码本身更有价值。

坑一：索引加错了顺序。 我一开始加的是 (created_at, user_id)，结果 EXPLAIN 还是 filesort。原因是联合索引遵循最左前缀，等值条件 user_id 必须在前，排序字段在后，才能既走索引又避免排序。这个顺序错了，索引基本白加。

坑二：joinedload 导致结果集膨胀。 我们有个接口是一对多（订单对商品），用 joinedload 后 SQL 返回了 20×N 行，反而更慢。换成 selectinload 后正常。教训是：一对一用 joinedload，一对多用 selectinload。

坑三：缓存 key 设计太粗。 一开始 key 是 orders:{user_id}，没带分页参数，导致不同页返回同一份数据。后来改成带 page:size，并且用 md5 压缩长度，避免 key 过长。

坑四：Gunicorn worker 数配错。 原来配的是 workers=8（CPU 核数 × 2），但服务是 4 核，8 个 worker 反而因为上下文切换导致性能下降。改成 workers=4，每个 worker 4 线程，QPS 提升了 15%。公式是 (2 × CPU) + 1，但这不是铁律，要实测。

坑五：压测时没关慢查询日志。 第一次压测数据很难看，后来发现是 MySQL 慢查询日志开着，每次写入都拖慢。关掉后数据才正常。压测环境要尽量贴近生产，但无关的日志、监控要关。

六、效果数据

压测用 locust，模拟 500 并发用户，持续 10 分钟，逐步加压。

指标 优化前 优化后 提升
QPS 320 2100 6.5x
P50 680ms 45ms 15x
P95 1800ms 120ms 15x
P99 3800ms 180ms 21x
超时率 12% 0.02% -
MySQL 慢查询/分钟 40+ 0 -
CPU 使用率 78% 42% -

分阶段看，每一步的收益：

  1. 索引优化：P99 从 3800ms → 1400ms，贡献最大，改动最小。
  2. 批量查询（消除 N+1）：1400ms → 620ms。
  3. Redis 缓存：620ms → 200ms。
  4. FastAPI 异步改造：网关层 30ms → 12ms，整体 P99 到 180ms。

这个顺序也印证了那个原则：先优化 SQL，再优化代码，最后上缓存。如果一上来就加缓存，索引问题被掩盖，数据量再涨还是会崩，而且缓存一失效就打回原形。

七、总结

这次调优下来，几点体会：

  1. profiling 是第一步，不是可选项。 没有 py-spy 和 cProfile，我可能会去猜是缓存问题或者网络问题，方向就错了。工具能让你看到真相。

  2. 索引是性价比最高的优化。 一条 ALTER TABLE 换来 60% 的性能提升，没有比这更划算的了。但要理解最左前缀和排序字段的位置。

  3. N+1 查询是隐形杀手。 单看每次查询都很快，但 20 次叠加就是 2.8 秒。ORM 用起来爽，但要警惕它的隐式查询。

  4. 缓存是最后手段，不是第一选择。 缓存引入一致性问题、穿透问题、雪崩问题，维护成本高。先把底层问题解决，缓存才能锦上添花。

  5. 异步框架里别写同步代码。 FastAPI 的 async 路由里跑 requests，等于把异步优势全扔了。这个坑很常见，但很容易被忽略。

  6. 压测数据要可信。 关掉无关日志、保证环境干净、逐步加压，不然数据会误导你。

最后贴一下优化后的核心代码，方便对照：

```python

order_service.py

from flask import Flask, request, jsonify
from sqlalchemy.orm import joinedload, selectinload
from cachetools import TTLCache
from redis import Redis
import json, hashlib

app = Flask(name)
local_cache = TTLCache(maxsize=10000, ttl=30)
redis_client = Redis(host="redis.internal", port=6379, db=0, socket_timeout=0.05)

def query_orders_from_db(user_id: int, page: int, size: int):
offset = (page - 1) * size
orders =