1. 问题背景：凌晨两点，订单API被爆了

上周五晚上线上告警，订单详情接口GET /api/v1/orders/{id}在凌晨大促流量下P99延迟从正常的120ms直接飙到3.8秒，数据库CPU持续100%运行，当时第一反应是“加机器”，但冷静下来知道这是典型的API性能瓶颈——不是硬件不够，而是代码和查询有问题。

我用的是FastAPI 0.100 + SQLAlchemy 2.0 + PostgreSQL 14，压测工具用wrk。另外一个同事的Flask 3.0项目也遇到类似问题，所以这篇文章会同时给出两个框架的对比方案。

2. 环境与版本：别再用老版本了

  • Python 3.11.4
  • FastAPI 0.100.0 / Flask 3.0.0
  • SQLAlchemy 2.0.19（1.4的query API已废弃）
  • Redis 6.2.6
  • PostgreSQL 14.5（max_connections=100）
  • wrk 4.2.0（压测工具）
  • py-spy 0.3.14（采样profiler）

压测机：8核16G，API和DB在同一台机器（避免网络干扰），Redis单独一台。

3. 第一步：Profiling —— 别猜，用py-spy直接看

我没有用cProfile那个老古董，py-spy的优势是不用改代码、不用重启进程，直接采样运行中的进程。

# 找到API进程PID
pgrep -f "uvicorn main:app"

# 采样10秒，输出火焰图
py-spy record --pid 15723 --output flamegraph.svg --duration 10

# 查看当前调用栈前20个热点
py-spy dump --pid 15723

火焰图显示72%的时间花在sqlalchemy.orm.loading._emit_load上，这就是典型的N+1问题。订单详情返回订单+订单项+商品+用户地址，每查询一个订单就额外发3条SQL，100个订单就是400条SQL，每条都要走网络 + 解析 + 执行。

另外用pg_stat_statements看了一下慢查询：

SELECT query, calls, mean_exec_time, rows FROM pg_stat_statements 
WHERE query LIKE '%orders%' ORDER BY mean_exec_time DESC LIMIT 5;

结果一条JOIN查询平均耗时2.4秒，原因是orders.user_id和order_items.order_id没有索引，全表扫描。

4. 第二步：数据库查询优化 —— 从400条SQL降到1条

原代码（FastAPI + SQLAlchemy 1.x风格）：

# 优化前：N+1查询
order = db.query(Order).filter(Order.id == order_id).first()
items = db.query(OrderItem).filter(OrderItem.order_id == order.id).all()
for item in items:
    product = db.query(Product).filter(Product.id == item.product_id).first()
    address = db.query(Address).filter(Address.user_id == order.user_id).first()

优化后（SQLAlchemy 2.0的selectinload，一次性批量加载关联对象）：

from sqlalchemy.orm import selectinload

# 优化后：单条JOIN查询 + selectin批量加载
async def get_order_detail(order_id: int):
    async with AsyncSessionLocal() as session:
        stmt = (
            select(Order)
            .options(
                selectinload(Order.items).selectinload(OrderItem.product),
                selectinload(Order.user).selectinload(User.address),
            )
            .where(Order.id == order_id)
        )
        result = await session.execute(stmt)
        return result.scalar_one()

加索引：

CREATE INDEX CONCURRENTLY idx_orders_user_id ON orders(user_id);
CREATE INDEX CONCURRENTLY idx_order_items_order_id ON order_items(order_id);
CREATE INDEX CONCURRENTLY idx_products_id ON products(id);

注意CONCURRENTLY会在生产环境不锁表，但需要非事务环境执行。

优化效果：数据库查询次数从400条降到3条（订单+订单项+商品/地址批量加载），平均查询时间从2.4秒降到85ms（EXPLAIN ANALYZE确认走了Index Scan）。

5. 第三步：缓存策略 —— Redis分级缓存，注意TTL和穿透

查询优化后P99降到800ms，但还是不够。发现热点商品sku（比如iPhone 15）被大量重复查询，数据库压力还是大。决定用Redis 6.2做两级缓存：

  • 第一级：商品详情缓存（TTL 300s，key: product:{sku_id}）
  • 第二级：订单详情缓存（TTL 60s，key: order:{id}:v2，订单状态变化时主动失效）

核心实现（FastAPI版本）：

import json
import redis.asyncio as aioredis

redis_client = aioredis.from_url(
    "redis://cache:6379/0",
    decode_responses=True,
    max_connections=50,
    socket_timeout=2,  # 防止redis拖垮API
)

async def get_order_cached(order_id: int):
    cache_key = f"order:{order_id}:v2"
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # 缓存未命中，查数据库
    order = await get_order_detail(order_id)
    if order:
        # 注意：主动失效逻辑在订单状态变更时删除key
        await redis_client.setex(cache_key, 60, json.dumps(order))
    return order

Flask 3.0版本类似，只是用flask_caching包，底层也是redis：

from flask_caching import Cache
cache = Cache(app, config={
    'CACHE_TYPE': 'redis',
    'CACHE_REDIS_URL': 'redis://cache:6379/0',
    'CACHE_DEFAULT_TIMEOUT': 300
})

@app.route('/api/v1/orders/')
@cache.cached(timeout=60, key_prefix='order:')
def get_order(order_id):
    # 原逻辑
    return jsonify(order_data)

6. 踩坑与优化：缓存穿透和连接池的坑

第一个坑：缓存穿透
压测时发现QPS 500后P99反而更高了，排查发现是个不存在的订单ID请求导致每次都要查数据库。解决方案：对不存在的订单也缓存空值（TTL 30s），或者在API入口加BloomFilter（我用pybloom_live实现）。

第二个坑：Redis连接池耗尽
当时把max_connections设成100，结果压测时Redis连接数打满，API直接报ConnectionError。后来调低到50，同时加了socket_timeout=2，让慢查询直接失败降级到数据库，而不是无限等待。

第三个坑：FastAPI的async和SQLAlchemy的selectinload不兼容
在FastAPI的异步路由里，selectinload必须搭配async_sessionmaker，不能用老的sessionmaker，否则会报MissingGreenlet错误。Flask没有这个问题因为它全同步。

7. 效果数据：压测对比

用wrk压测（持续60秒，16线程，连接数64，GET请求带真实订单ID）：

场景 QPS P99延迟 数据库CPU Redis QPS
优化前（原始代码） 120 3.8s 100% 0
优化后（加索引+selectinload） 420 780ms 55% 0
优化+Redis缓存（TTL 300s） 850 180ms 15% 1500

压测命令：

wrk -t16 -c64 -d60s --latency -s post.lua http://localhost:8000/api/v1/orders/123456

最终总结：API性能调优的优先级是：先看Profile火焰图 → 优化数据库查询（加索引+减少查询次数） → 加缓存（但注意穿透和连接池）。不要一开始就上缓存，缓存掩盖了真的问题——慢查询。另外，FastAPI的异步在IO密集型场景比Flask强（压测反压测，FastAPI在优化后QPS比Flask高约15%，因为异步非阻塞更适合并发），但如果你的代码是同步的，Flask 3.0加gunicorn多worker也能扛住，只是资源消耗更高。

最后留个问题：如果你的缓存TTL是60s，但订单状态在10s内变更了，用户会看到过期数据。我用的是主动失效（订单状态变更时deleted key），如果你用纯TTL，业务上能接受吗？评论区聊。