FastAPI与Flask压测对比：PySpy定位慢查询与Redis缓存层优化实录

# 安装
pip install py-spy
# 抓取10秒火焰图，导出为svg
py-spy record --pid 12345 --duration 10 --format flamegraph -o /tmp/order_api.svg
# 同时开启PostgreSQL慢查询日志
vim /etc/postgresql/15/main/postgresql.conf
log_min_duration_statement = 200ms
log_line_prefix = '%t [%p]: '

[2023-11-02 14:31:22] [12345]: duration: 850ms execute SELECT * FROM orders WHERE id = $1
[2023-11-02 14:31:22] [12346]: duration: 830ms execute SELECT * FROM order_items WHERE order_id = $1
[2023-11-02 14:31:22] [12347]: duration: 780ms execute SELECT * FROM users WHERE id = $1

# Flask + SQLAlchemy 懒加载写法（问题代码）
@app.route('/api/orders/')
def get_order(order_id):
    order = db.session.query(Order).filter(Order.id == order_id).one()
    # 每次访问 order.items 都会触发一条新SQL
    items = [{
        'id': item.id,
        'product_name': item.product.name,  # 懒加载 product
        'price': item.price
    } for item in order.items]
    # user 也是懒加载，而且用了子查询计算总额
    user = order.user
    total = db.session.query(func.sum(OrderItem.price)).filter(OrderItem.order_id == order_id).scalar()
    return jsonify({'order': order.to_dict(), 'items': items, 'user': user.name, 'total': total})

# 优化后：SQLAlchemy 2.0 selectinload
from sqlalchemy.orm import selectinload, joinedload
from sqlalchemy import func, select

@app.route('/api/orders/')
def get_order(order_id):
    stmt = (
        select(Order)
        .options(
            selectinload(Order.items).selectinload(OrderItem.product),
            joinedload(Order.user),
            selectinload(Order.shipping_address)
        )
        .where(Order.id == order_id)
    )
    order = db.session.execute(stmt).scalar_one()

    # 总额用Python累加，不再单独查库
    total = sum(item.price for item in order.items)

    return jsonify({
        'order_id': order.id,
        'user': order.user.name,
        'items': [{'id': i.id, 'product': i.product.name, 'price': i.price} for i in order.items],
        'total': round(total, 2)
    })

# 缓存层代码（Redis + 手动序列化）
import redis
import pickle

redis_client = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=False)
CACHE_TTL = 300  # 5分钟

@app.route('/api/orders/')
def get_order_cached(order_id):
    cache_key = f'order:detail:{order_id}'
    # 第一级：Redis 读
    cached = redis_client.get(cache_key)
    if cached:
        return jsonify(pickle.loads(cached))

    # 第二级：查库（经过预加载优化后的查询）
    stmt = (
        select(Order)
        .options(
            selectinload(Order.items).selectinload(OrderItem.product),
            joinedload(Order.user),
            selectinload(Order.shipping_address)
        )
        .where(Order.id == order_id)
    )
    order = db.session.execute(stmt).scalar_one()

    data = {
        'order_id': order.id,
        'user': order.user.name,
        'items': [{'id': i.id, 'product': i.product.name, 'price': i.price} for i in order.items],
        'total': round(sum(i.price for i in order.items), 2)
    }
    # 写入缓存，用pickle保留float精度，避免json.dumps丢失
    redis_client.setex(cache_key, CACHE_TTL, pickle.dumps(data))
    return jsonify(data)

# FastAPI 异步路由
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import asyncpg

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/db"
engine = create_async_engine(DATABASE_URL, pool_size=10, max_overflow=20)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

app = FastAPI()

@app.get("/api/orders/{order_id}")
async def get_order(order_id: int):
    async with AsyncSessionLocal() as session:
        stmt = (
            select(Order)
            .options(selectinload(Order.items).selectinload(OrderItem.product))
            .where(Order.id == order_id)
        )
        result = await session.execute(stmt)
        order = result.scalar_one()
        return {'order_id': order.id, 'items': [{'id': i.id} for i in order.items]}

# gunicorn.conf.py
workers = 4  # 4个进程
worker_class = "uvicorn.workers.UvicornWorker"
threads = 1  # 异步worker不需要线程
timeout = 30