一、问题的起源：一次线上告警

凌晨2点，钉钉告警：订单查询接口P99超过4秒。这个接口在双11大促期间承担了60%的读流量，负责聚合用户信息、订单列表、商品快照三个微服务的数据。最初版本用Flask开发，后来用FastAPI重写了部分逻辑，但性能并没有预期的提升——平均响应时间从1.2s反而涨到了1.8s。

直觉告诉我：问题不在框架本身，而在数据访问层。但在我用time.time()打点排查之前，一切都是猜测。这篇文章记录了我从零开始定位和优化的全过程，包括两次翻车经历和最终的性能数据。

二、环境与版本说明

  • 服务框架：FastAPI 0.104.1（异步路由）+ Flask 3.0.0（同步路由）双版本对比
  • ASGI服务器：Uvicorn 0.24.0（workers=4）
  • WSGI服务器：Gunicorn 21.2.0（workers=4, threads=8）
  • ORM：SQLAlchemy 2.0.25 + asyncpg 0.29.0（FastAPI）/ PyMySQL 1.1.0（Flask）
  • 缓存：Redis 5.0.3（单节点，maxmemory-policy allkeys-lru）
  • 压测工具：wrk 4.2.0，测试场景：200并发，持续60秒
  • 数据量：用户表50万行，订单表200万行，商品快照表100万行

三、定位瓶颈：别猜，用工具说话

3.1 第一轮：cProfile快速扫描

写了个脚本直接压测并抓取profile数据：

# perf_test.py
import cProfile
import pstats
import io
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def run_benchmark():
    for _ in range(500):
        client.get("/api/v1/orders?user_id=10001&page=1&size=20")

pr = cProfile.Profile()
pr.enable()
run_benchmark()
pr.disable()

s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(30)
print(s.getvalue())

输出关键片段：

ncalls  tottime  cumtime  filename:lineno(function)
500     12.345   45.678   asyncpg/protocol/protocol.pyx:1(asyncpg)
500     8.234    32.456   sqlalchemy/orm/loading.py:1(load_on_ident)
2500    5.678    28.901   sqlalchemy/orm/relationships.py:1(_selectin_loader)

问题暴露了：2500次relationship加载。一个订单列表接口，SQLAlchemy默认的lazy loading在每次访问order.items时都发一条SQL。200万订单数据导致N+1查询，总共发了2500+次数据库往返。

3.2 第二轮：py-spy盯生产环境

cProfile有性能开销（大约20%），不适合直接打生产。用py-spy dump抓线程栈：

py-spy dump --pid 12345

发现大量线程阻塞在await session.commit()——这是另一个坑：FastAPI的异步Session在每次请求结束时都执行隐式flush+commit，而我们的场景是纯只读接口，这完全是浪费。

四、方案设计：三层缓存 + 查询重构

4.1 缓存架构设计

┌─────────────┐
│  API Layer  │  ← FastAPI/Flask 路由处理
└──────┬──────┘
       │
┌──────▼──────┐
│  Cache L1   │  ← 进程内缓存 (ttl=5s, maxsize=1024)
└──────┬──────┘
       │
┌──────▼──────┐
│  Cache L2   │  ← Redis (ttl=60s, key: prefix:user_id:page:size)
└──────┬──────┘
       │
┌──────▼──────┐
│   DB Layer  │  ← SQLAlchemy + selectinload
└─────────────┘
  • L1：本地内存缓存，用cachetools的TTLCache，解决热点数据重复查询
  • L2：Redis缓存，key设计为orders:{user_id}:{page}:{size}，value为JSON序列化后的完整响应
  • L3：数据库查询优化，消除N+1

4.2 数据库查询优化：selectinload

# FastAPI版本 - 使用selectinload避免N+1
from sqlalchemy.orm import selectinload
from sqlalchemy import select

async def get_orders_with_items(user_id: int, page: int, size: int):
    stmt = (
        select(Order)
        .options(
            selectinload(Order.items),  # 一次IN查询加载所有items
            selectinload(Order.user).joinedload(User.profile)  # 嵌套加载
        )
        .where(Order.user_id == user_id)
        .offset((page - 1) * size)
        .limit(size)
    )
    result = await db.execute(stmt)
    return result.scalars().unique().all()

注意两个关键点：
1. selectinload会生成WHERE id IN (...)查询，将2500次查询降至2次
2. unique()必须调用，否则FastAPI的异步session会报错

4.3 缓存装饰器实现

# cache_decorator.py
import json
import hashlib
from functools import wraps
from cachetools import TTLCache
import redis.asyncio as redis

# L1 进程内缓存
local_cache = TTLCache(maxsize=1024, ttl=5)

# L2 Redis缓存
redis_client = redis.from_url("redis://localhost:6379/0", decode_responses=True)

def cached_api(prefix: str, ttl: int = 60):
    """三级缓存装饰器：本地 -> Redis -> 原始函数"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 构建缓存key
            raw_key = f"{prefix}:{args}:{kwargs}"
            cache_key = hashlib.md5(raw_key.encode()).hexdigest()

            # L1 查本地
            local_data = local_cache.get(cache_key)
            if local_data:
                return json.loads(local_data)

            # L2 查Redis
            redis_data = await redis_client.get(cache_key)
            if redis_data:
                local_cache[cache_key] = redis_data  # 回填本地
                return json.loads(redis_data)

            # L3 执行原始函数
            result = await func(*args, **kwargs)
            serialized = json.dumps(result, default=str)

            # 写入两级缓存
            local_cache[cache_key] = serialized
            await redis_client.setex(cache_key, ttl, serialized)

            return result
        return wrapper
    return decorator

# 使用示例
@cached_api(prefix="orders", ttl=60)
async def get_orders(user_id: int, page: int, size: int):
    # 数据库查询逻辑
    ...

4.4 Flask版本对比

Flask是同步框架，缓存装饰器需要处理线程安全：

# flask_cache.py
import threading
from cachetools import TTLCache
import redis

local_cache = TTLCache(maxsize=1024, ttl=5)
cache_lock = threading.Lock()
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cached_api_flask(prefix: str, ttl: int = 60):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = f"{prefix}:{args}:{kwargs}"
            with cache_lock:
                local_data = local_cache.get(cache_key)
            if local_data:
                return local_data

            redis_data = redis_client.get(cache_key)
            if redis_data:
                with cache_lock:
                    local_cache[cache_key] = redis_data
                return json.loads(redis_data)

            result = func(*args, **kwargs)
            serialized = json.dumps(result, default=str)
            with cache_lock:
                local_cache[cache_key] = serialized
            redis_client.setex(cache_key, ttl, serialized)
            return result
        return wrapper
    return decorator

Flask版本必须加threading.Lock()保护本地缓存，因为Gunicorn的threads=8意味着8个线程并发访问。这是一个容易踩的坑：不加锁会导致TTLCache内部状态损坏，偶尔出现KeyError。

五、踩坑与优化记录

5.1 坑1：FastAPI异步Session的隐式commit

现象：压测时发现数据库连接数飙升到200+。

原因：FastAPI的async_sessionmaker默认expire_on_commit=True，每次请求结束自动commit，触发一次额外的DB往返。

解决：

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
)

async_session = async_sessionmaker(
    engine,
    expire_on_commit=False,  # 关键：关闭自动过期
    autoflush=False,          # 关闭自动flush
    class_=AsyncSession
)

5.2 坑2：Redis连接池耗尽

现象：压测后期出现ConnectionError: Error while reading from socket。

原因：Redis客户端默认连接池大小只有50，而压测并发200。

解决：

redis_client = redis.from_url(
    "redis://localhost:6379/0",
    decode_responses=True,
    max_connections=200,
    socket_timeout=2,
    socket_connect_timeout=2,
)

5.3 坑3：本地缓存与Redis的TTL不一致

现象：数据更新后，本地缓存还在生效（TTL=5s），导致Redis的60s缓存失效后仍然返回旧数据。

解决：本地缓存TTL必须小于Redis TTL，我这里设置5s < 60s，保证最终一致性。如果要求更高，需要引入消息总线主动失效。

六、压测数据对比

使用wrk压测，命令：

wrk -t8 -c200 -d60s --latency http://localhost:8000/api/v1/orders?user_id=10001&page=1&size=20

6.1 原始版本（Flask 3.0.0）

指标 数值
平均延迟 1800ms
P99延迟 4200ms
QPS 110 req/s
数据库查询数/请求 52次
CPU占用 85%

6.2 优化后FastAPI + 三级缓存

指标 数值
平均延迟 220ms（↓87.8%）
P99延迟 380ms（↓91%）
QPS 455 req/s（↑313%）
数据库查询数/请求 2次
CPU占用 45%

6.3 优化后Flask + 三级缓存

指标 数值
平均延迟 350ms（↓80.5%）
P99延迟 650ms（↓84.5%）
QPS 285 req/s（↑159%）
数据库查询数/请求 2次
CPU占用 60%

6.4 关键发现

FastAPI在优化后性能优势更明显：因为异步路由配合asyncpg驱动，数据库I/O可以并发执行。而Flask的同步模型下，即使查询次数相同，也无法重叠数据库等待时间。

七、部署调参建议

7.1 Gunicorn + Uvicorn 配置

# gunicorn.conf.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
bind = "0.0.0.0:8000"
timeout = 30
graceful_timeout = 10
keepalive = 5
max_requests = 1000  # 防止内存泄漏
max_requests_jitter = 100

7.2 Redis内存策略

# redis.conf
maxmemory 512mb
maxmemory-policy allkeys-lru  # 淘汰策略，缓存不是强一致要求
save ""  # 关闭RDB持久化，纯缓存场景
appendonly no

八、总结与反思

这次调优的核心收获：

  1. 别猜，用工具。cProfile和py-spy五分钟就定位了N+1问题，比肉眼review代码高效十倍。
  2. 框架不是瓶颈。FastAPI和Flask的框架开销差异在10%以内，真正的性能杀手在ORM使用方式和缓存策略。
  3. 缓存要分三级。本地缓存扛热点，Redis扛分布式一致性，数据库兜底。每级的TTL必须精心设计。
  4. 异步不是银弹。Flask同步模型配合线程池也能达到285 QPS，对于中小型项目完全够用。

最后给一个建议：如果你的接口P99超过1s，先不要急着换框架或者加机器。用py-spy dump --pid抓几个线程栈看看，大概率是数据库查询慢或者N+1问题。优化数据访问层，通常能带来80%以上的性能提升。