FastAPI与Flask API性能调优：从10秒到200ms的瓶颈拆解

# profiling_middleware.py
import cProfile
import io
import pstats
from functools import wraps
from fastapi import Request, Response
import time

def profile_endpoint(func):
    @wraps(func)
    async def wrapper(request: Request, *args, **kwargs):
        pr = cProfile.Profile()
        pr.enable()
        start = time.perf_counter()
        response = await func(request, *args, **kwargs)
        elapsed = time.perf_counter() - start
        pr.disable()

        s = io.StringIO()
        ps = pstats.Stats(pr, stream=s).sort_stats('cumtime')
        ps.print_stats(20)  # 只打印前20耗时

        # 写入日志或返回header
        print(f"Endpoint {request.url.path} took {elapsed:.3f}s")
        print(s.getvalue()[:2000])  # 截断输出
        return response
    return wrapper

# 使用方式：在路由上 @router.get("/orders")(profile_endpoint)

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool

# 关键配置：连接池大小20，溢出10，回收300秒
engine = create_engine(
    "mysql+pymysql://user:pass@host:3306/orders",
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=10,
    pool_recycle=300,
    pool_pre_ping=True
)
SessionLocal = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

# order_service.py
from sqlalchemy.orm import joinedload
from models import Order, User

def get_orders_with_user(db, limit=20, offset=0):
    # 之前写法：先查订单，再循环查用户
    # orders = db.query(Order).offset(offset).limit(limit).all()
    # 优化后：一次性join加载user
    orders = db.query(Order).options(
        joinedload(Order.user)  # 假设Order模型有user关系
    ).offset(offset).limit(limit).all()

    # 直接访问Order.user不再触发额外查询
    result = []
    for order in orders:
        result.append({
            "order_id": order.id,
            "amount": order.amount,
            "user_name": order.user.name  # 懒加载已预填充
        })
    return result

# main.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import ujson
import datetime

app = FastAPI()

def custom_serializer(obj):
    if isinstance(obj, datetime.datetime):
        return obj.isoformat()
    raise TypeError(f"Type {type(obj)} not serializable")

def json_streamer(data_generator):
    """逐块生成JSON数组，避免一次性构建大字符串"""
    yield "["
    first = True
    for item in data_generator:
        if not first:
            yield ","
        first = False
        # 使用ujson.dumps，速度比标准json快3-5倍
        yield ujson.dumps(item, default=custom_serializer)
    yield "]"

@app.get("/orders/stream")
def get_orders_stream(page: int = 1, page_size: int = 20):
    db = SessionLocal()
    try:
        orders = get_orders_optimized(db, page, page_size)
        # 生成器逐条返回
        def gen():
            for o in orders:
                yield {
                    "order_id": o.id,
                    "amount": float(o.amount),
                    "user_name": o.user.name,
                    "created_at": o.created_at
                }
        return StreamingResponse(
            json_streamer(gen()),
            media_type="application/json"
        )
    finally:
        db.close()

# cache_middleware.py
import hashlib
import json
import redis.asyncio as aioredis
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

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

class RedisCacheMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 只缓存GET请求
        if request.method != "GET":
            return await call_next(request)

        # 生成缓存key（基于路径+查询参数）
        cache_key = f"api_cache:{request.url.path}:{hashlib.md5(json.dumps(dict(request.query_params)).encode()).hexdigest()}"

        # 尝试从Redis获取
        cached = await redis_client.get(cache_key)
        if cached:
            return Response(content=cached, media_type="application/json")

        # 执行真实请求
        response = await call_next(request)

        # 写入Redis，TTL 60秒
        if response.status_code == 200:
            body = b""
            async for chunk in response.body_iterator:
                body += chunk
            await redis_client.setex(cache_key, 60, body.decode())
            # 重新构造Response
            return Response(content=body, media_type=response.media_type, status_code=response.status_code)

        return response