FastAPI与Flask性能对决：Profile驱动调优从2s到120ms

Python 3.11.4
Flask 2.3.2 / FastAPI 0.100.0
SQLAlchemy 1.4.48 (asyncpg 0.27.0)
PostgreSQL 14.5 (shared_buffers=4GB, work_mem=64MB)
Redis 7.0 (maxmemory-policy=allkeys-lru)
gunicorn 21.2.0 (worker_class=gevent / uvicorn)
wrk 4.2.0 (压测工具)

# app_flask_optimized.py
from flask import Flask, jsonify, request
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, selectinload
import redis, json, time

app = Flask(__name__)
engine = create_engine(
    'postgresql+psycopg2://user:pass@localhost/orders',
    pool_size=20,          # 默认5，改到20
    max_overflow=10,       # 最多额外10个连接
    pool_pre_ping=True,    # 防止连接失效
    pool_recycle=300       # 5分钟回收
)
Session = sessionmaker(bind=engine)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

@app.route('/orders')
def get_orders():
    user_id = request.args.get('user_id', type=int)
    if not user_id:
        return jsonify({'error': 'missing user_id'}), 400

    # 缓存策略：先查Redis，命中直接返回
    cache_key = f'order:list:{user_id}'
    cached = r.get(cache_key)
    if cached:
        return jsonify(json.loads(cached))

    # 数据库查询：一次性JOIN，避免N+1
    with Session() as s:
        stmt = text("""
            SELECT o.id, o.amount, p.name AS product_name, s.name AS shop_name
            FROM orders o
            JOIN products p ON o.product_id = p.id
            JOIN shops s ON o.shop_id = s.id
            WHERE o.user_id = :uid
            ORDER BY o.created_at DESC
            LIMIT 20
        """)
        result = s.execute(stmt, {'uid': user_id}).mappings().all()
        data = [dict(row) for row in result]

    # 回填缓存，60秒过期
    r.setex(cache_key, 60, json.dumps(data))
    return jsonify(data)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000)

# app_fastapi.py
from fastapi import FastAPI, Query
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text
import redis.asyncio as aioredis
import json

app = FastAPI()
# asyncpg驱动，注意URL前缀是postgresql+asyncpg
engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/orders',
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True
)
SessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
r = aioredis.from_url('redis://localhost:6379/0', decode_responses=True)

@app.get('/orders')
async def get_orders(user_id: int = Query(...)):
    cache_key = f'order:list:{user_id}'
    cached = await r.get(cache_key)
    if cached:
        return json.loads(cached)

    async with SessionLocal() as s:
        result = await s.execute(text("""
            SELECT o.id, o.amount, p.name AS product_name, s.name AS shop_name
            FROM orders o
            JOIN products p ON o.product_id = p.id
            JOIN shops s ON o.shop_id = s.id
            WHERE o.user_id = :uid
            ORDER BY o.created_at DESC
            LIMIT 20
        """), {'uid': user_id})
        rows = result.mappings().all()
        data = [dict(row) for row in rows]

    await r.setex(cache_key, 60, json.dumps(data))
    return data