FastAPI与Flask性能对决：Profiling驱动DB查询与Redis缓存优化实录

1. 问题背景：线上订单接口的“雪崩”预警

接手的是一个用FastAPI(0.95.1)写核心API、Flask(2.2.3)写管理后台的混合项目。某天监控告警：/api/v1/orders接口的P95延迟从正常300ms飙至1.8s，QPS从500跌至150。查看Grafana，数据库连接池打满，CPU使用率100%。

初步排查发现：订单查询接口会返回用户信息、商品列表、物流状态，而ORM用的是SQLAlchemy 2.0.19。直觉告诉我——大概率是N+1查询。但这次我决定用工具说话，不再靠猜。

2. 环境与版本：先交代底细

• Python 3.10.12
• FastAPI 0.95.1 + Uvicorn 0.23.1 (worker=4)
• Flask 2.2.3 + Gunicorn 20.1.0 (worker=4, sync)
• SQLAlchemy 2.0.19 + psycopg2-binary 2.9.9
• Redis 7.0.11 (python客户端: redis-py 4.5.4)
• PostgreSQL 14.5 (配置: max_connections=100, shared_buffers=256MB)
• 压测工具: wrk 4.2.0 / 单机8核16G

3. Profiling定位：别猜，用数据说话

先用py-spy(0.3.14)对运行中的FastAPI进程进行采样：

# 安装与采样
pip install py-spy
sudo py-spy record --pid 12345 -o profile.svg --duration 30

生成的火焰图清晰显示：SQLAlchemy的fetchall()和lazy_load()占据67%的CPU时间。再结合cProfile做函数级分析：

import cProfile
import pstats
from app.api.orders import get_orders

profiler = cProfile.Profile()
profiler.enable()
# 模拟实际调用，传入100个订单ID
get_orders(limit=100)
profiler.disable()
stats = pstats.Stats(profiler).sort_stats('cumulative')
stats.print_stats(20)

输出关键行：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
103     0.012    0.000    0.542    0.005  sqlalchemy/orm/loading.py:322(load_scalar_attribute)
1       0.008    0.008    1.203    1.203  app/services/order_service.py:45(get_orders)

103次标量属性加载，累计0.542秒——这就是N+1的实锤。每个订单查询用户、商品、物流，各发一次查询。

4. 方案设计：三管齐下

方案一：SQLAlchemy查询优化 — 使用joinedload或selectinload一次性加载关联对象。

方案二：Redis二级缓存 — 对用户信息和商品详情做缓存，key设计为user:{id}、product:{id}，TTL设为300秒。

方案三：异步化改造 — FastAPI部分用async def + asyncpg，避免线程阻塞；Flask管理后台暂时保留同步，但把模板渲染交给render_template自带的缓存。

5. 核心实现：代码与踩坑

第一步：修复N+1查询

# app/api/orders.py - 优化前
from sqlalchemy.orm import Session
from app.models import Order, User, Product

def get_orders(db: Session, limit: int):
    orders = db.query(Order).limit(limit).all()
    result = []
    for order in orders:
        user = db.query(User).filter(User.id == order.user_id).first()  # N次查询
        products = db.query(Product).filter(Product.id.in_(order.product_ids)).all()  # N次查询
        result.append({
            "order_id": order.id,
            "user": {"name": user.name, "email": user.email},
            "products": [p.name for p in products]
        })
    return result

# 优化后 - 使用selectinload
from sqlalchemy.orm import selectinload

def get_orders_optimized(db: Session, limit: int):
    orders = db.query(Order).options(
        selectinload(Order.user),
        selectinload(Order.products)
    ).limit(limit).all()
    return [
        {
            "order_id": o.id,
            "user": {"name": o.user.name, "email": o.user.email},
            "products": [p.name for p in o.products]
        }
        for o in orders
    ]

踩坑1：joinedload在SQLAlchemy 2.0中对limit分页会生成子查询，导致性能反而下降。selectinload更安全，它发出第二条IN查询，但总查询数从2N+1降到3条。

第二步：Redis缓存层

# app/cache.py
import json
import redis.asyncio as aioredis

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

async def get_or_cache(key: str, fetch_func, ttl: int = 300):
    cached = await redis_client.get(key)
    if cached:
        return json.loads(cached)
    data = await fetch_func()
    await redis_client.setex(key, ttl, json.dumps(data))
    return data

踩坑2：使用异步Redis客户端时必须注意FastAPI的async def与同步SQLAlchemy的兼容性。SQLAlchemy 2.0的Session是线程安全的，但在异步函数中调用同步DB操作会阻塞事件循环。解决方案：用run_in_executor或直接用asyncpg重写DB层。我选择了前者，改动最小：

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=8)

async def get_orders_async(db: Session, limit: int):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, get_orders_optimized, db, limit)

第三步：压测对比

用wrk进行3轮压测，每轮60秒，100并发连接：

wrk -t8 -c100 -d60s --latency http://localhost:8000/api/v1/orders?limit=50

6. 效果数据：从惨不忍睹到舒服了

关键数据细节：
- 优化前：每条订单查询产生4条SQL（订单主查询 + 用户 + 商品列表 + 物流），100条订单=1+3*100=301条SQL。
- 优化后：selectinload将SQL降到3条（订单 + 用户IN查询 + 商品IN查询），同时Redis命中率92%时，DB查询量再降60%。
- 缓存命中率对QPS影响显著：命中率从0到92%，QPS从420（仅SQL优化）提升到2100。说明缓存是决定性因素。

7. 踩坑与优化：那些文档没告诉你的

坑1：SQLAlchemy 2.0的selectinload分页陷阱
实测joinedload+limit在PostgreSQL上生成LEFT JOIN子查询，导致行数膨胀，性能比N+1还差。必须用selectinload。

坑2：FastAPI的async def与同步DB混用
即使加了run_in_executor，频繁切换也会增加开销。最佳实践：新代码用asyncpg + SQLAlchemy Core异步版本，旧代码逐步迁移。

坑3：Redis缓存雪崩
初始TTL设置600秒，导致凌晨3点大量key同时过期，DB瞬间被打爆。后来TTL加入随机偏移量ttl = 300 + random.randint(0, 60)。

坑4：Flask管理后台的同步渲染
Flask端render_template对订单详情页每次请求都重查数据库，压测QPS只有120。改为@cache.cached(timeout=60)后，QPS到800，但页面数据最多滞后60秒——对管理后台可接受。

最终架构：
- FastAPI层：async def + run_in_executor处理同步DB，Redis缓存用户/商品。
- Flask层：flask-caching 3.0.2的Cache装饰器，缓存整个HTML片段。
- 数据库层：PgBouncer连接池（transaction模式），防止连接数打满。

8. 总结：性能调优的“道与术”

这次调优最大的教训是：先profiling再动手，90%的性能问题不需要猜。py-spy的火焰图直接暴露了N+1查询，而SQLAlchemy的优化是基础，Redis缓存是杠杆——两者叠加才实现量变到质变。

另外，不要迷信异步框架。FastAPI的异步特性只在IO密集型任务中发挥优势，如果DB层是同步的，性能提升有限。真正的瓶颈在数据库查询次数和缓存策略。

最后，压测数据要留档。我保留了每次优化后的wrk输出，对比时才能知道哪一步真正有效。建议你用同样的方法，在自己项目上试试——欢迎在评论区交流你的性能调优故事。

相关工具版本：py-spy 0.3.14, cProfile (Python标准库), wrk 4.2.0, SQLAlchemy 2.0.19, FastAPI 0.95.1, Flask 2.2.3, Redis 7.0.11。