1. 问题背景：上线当天被运维喊去喝茶

事情是这样的，我们有个订单查询服务，用FastAPI写的，部署在4核8G的容器里。上线当天，运维直接甩来一张监控图：P99延迟从200ms飙到3.2s，CPU直接跑满100%。我第一反应是"数据库连接池爆了？"，结果一查，连接池才用了30%。后来用cProfile一跑，发现60%的时间花在sqlalchemy.orm.query.Query.__iter__上——典型的懒加载N+1问题。

更打脸的是，同组的老哥用Flask写了个类似接口，虽然QPS更低（650），但人家延迟稳定啊。这让我意识到，框架快不等于接口快，性能瓶颈往往藏在ORM和序列化里。

2. 环境与版本：所有数据可复现

Python 3.10.12
FastAPI 0.104.1 + Uvicorn 0.24.0
Flask 2.3.3 + Gunicorn 21.2.0 (gevent worker)
SQLAlchemy 2.0.23
PostgreSQL 14.5 (max_connections=200, shared_buffers=2GB)
Redis 7.0.12
压测工具: wrk 4.2.0 (8线程, 200连接, 60s)

服务器：4核Intel Xeon 8374C, 8GB RAM, SSD磁盘。所有压测均在独立测试环境进行，排除网络波动。

3. 方案设计：三层递进优化

第一层：Profiling定位 —— 用cProfile静态分析 + py-spy动态抓取。
第二层：数据库查询优化 —— 解决N+1、减少往返次数。
第三层：缓存策略 —— 对热数据做Redis缓存，对冷数据用LRU本地缓存。

整体思路：先让查询"瘦身"，再加缓存"兜底"，最后根据压测数据决定是否上异步。

4. 核心实现：从定位到重构

4.1 Profiling工具使用实录

先用cProfile跑一个单次请求：

import cProfile
import pstats
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def make_request():
    for _ in range(100):
        client.get("/orders?user_id=12345")

profiler = cProfile.Profile()
profiler.enable()
make_request()
profiler.disable()

stats = pstats.Stats(profiler)
stats.sort_stats('cumulative').print_stats(20)

输出关键行：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
100    0.002    0.000    18.432    0.184  sqlalchemy/orm/loading.py:126(load_scalar_attribute)

坑1：load_scalar_attribute累计耗时18.4s，这就是懒加载的锅。每个订单对象关联的user和product子查询，100个请求产生了2500+条SQL。

接着用py-spy抓线上栈：

py-spy dump --pid 12345

看到多个线程卡在psycopg2.extensions.cursor.execute，确认是数据库等待。

4.2 数据库查询优化：干掉N+1

方案：用selectinload替代懒加载，同时把分页查询的count(*)优化为row_number() over()窗口函数。

# 优化前：懒加载，每个订单触发2次额外查询
orders = db.query(Order).filter(Order.user_id == user_id).offset(0).limit(20).all()

# 优化后：一次性关联加载
from sqlalchemy.orm import selectinload
orders = (
    db.query(Order)
    .options(selectinload(Order.items).selectinload(Item.product))
    .filter(Order.user_id == user_id)
    .offset(0)
    .limit(20)
    .all()
)

坑2：selectinload对一对多关系会生成IN查询，但如果关联表数据量大，IN列表可能超过PostgreSQL的max_parameters（默认32767）。我们订单详情只有20条，完全没问题，但如果你做批量导出，记得分段。

效果：SQL数量从2500+降到200（20个请求×10个关联查询），单请求数据库耗时从180ms降到32ms。

4.3 缓存策略：Redis + 本地LRU双层缓存

方案：对user_id维度的订单列表做缓存，TTL=300s。用Redis存JSON序列化后的数据，本地用functools.lru_cache兜底（应对Redis抖动）。

import json
import redis
from functools import lru_cache

redis_client = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

CACHE_TTL = 300
CACHE_PREFIX = "orders:v1"

@lru_cache(maxsize=256)  # 本地热缓存，最多保留256个user的订单
def get_orders_from_local_cache(user_id: int) -> str:
    return ""

async def get_orders_cached(user_id: int):
    cache_key = f"{CACHE_PREFIX}:{user_id}"

    # 先查本地LRU
    local_val = get_orders_from_local_cache(user_id)
    if local_val:
        return json.loads(local_val)

    # 再查Redis
    redis_val = redis_client.get(cache_key)
    if redis_val:
        get_orders_from_local_cache.cache_clear()  # 简单粗暴更新本地
        get_orders_from_local_cache.__wrapped__.__defaults__ = (redis_val,)
        return json.loads(redis_val)

    # 查询数据库并写缓存
    orders = fetch_orders_from_db(user_id)
    redis_client.setex(cache_key, CACHE_TTL, json.dumps(orders))
    return orders

坑3：lru_cache不能直接动态更新值，上面代码里用了__defaults__这种hack，实际生产建议用cachetools.TTLCache替代，支持显式更新。我这里为了演示简单，但真实项目别这么写。

效果：命中缓存时，接口耗时从32ms降到2ms。压测数据里，缓存命中率约65%（因为用户分布集中），整体QPS提升明显。

4.4 异步改造（仅FastAPI）

FastAPI天然支持async def，但SQLAlchemy是同步的，直接await会阻塞事件循环。我们改用asyncpg + SQLAlchemy 2.0的异步扩展：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

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

async_session = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def get_orders_async(user_id: int):
    async with async_session() as session:
        result = await session.execute(
            select(Order).options(selectinload(Order.items)).where(Order.user_id == user_id)
        )
        return result.scalars().all()

坑4：asyncpg默认不支持%占位符，SQLAlchemy会自动转成$1，但如果你的SQL里有LIKE '%foo%'，必须显式用text()并传参，否则报错。

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

  1. Gunicorn worker类型：Flask用gevent worker比sync快3倍，但必须设置worker_connections大于连接数。我们用了gunicorn -w 4 -k gevent -b 0.0.0.0:8000，但发现gevent monkey-patching和psycopg2冲突，必须用psycopg2.pool且手动monkey.patch_all()。

  2. JSON序列化：FastAPI默认用json.dumps，但Pydantic v2的序列化比v1快5倍。升级后单次序列化从1.8ms降到0.35ms。Flask则建议用orjson替代内置json，实测快8倍。

  3. 连接池参数：PostgreSQL的max_connections是200，但SQLAlchemy连接池默认pool_size=5，压测时大量连接等待。改成pool_size=20, max_overflow=10，QPS直接翻倍。

  4. 缓存穿透：某次压测发现Redis里不存在user_id=99999，每请求都打DB。加了None缓存（TTL=60s），数据库压力降了40%。

6. 效果数据：调优前后对比

指标 FastAPI优化前 FastAPI优化后 Flask优化前 Flask优化后
QPS 1800 9200 650 4100
P99延迟 1200ms 85ms 2400ms 210ms
数据库QPS 4500 900 3500 700
CPU平均占用 98% 42% 99% 55%
内存平均占用 800MB 1.2GB 700MB 1.1GB

解读：缓存和查询优化把数据库压力降了80%，但CPU占用反而升了（因为序列化更高效但更耗CPU）。内存升了是因为Redis客户端和本地缓存。Flask提升比FastAPI小，主要因为GIL限制，但加了gevent后多核利用率仍不如FastAPI的异步。

7. 总结：没有银弹，只有定位

这次调优最大的收获是：不要迷信框架性能对比。FastAPI号称比Flask快，但实际瓶颈在ORM和序列化。如果你遇到类似问题，按这个顺序排查：

  1. 先上cProfile看CPU时间分布，别猜。
  2. 用py-spy抓线上栈，确认是不是数据库等待。
  3. SQL优化优先级高于缓存：先干掉N+1，再考虑Redis。
  4. 缓存一定要考虑穿透和雪崩，TTL加随机值。
  5. 最后才考虑异步，因为异步改造代码变动大，且对Flask收益有限。

后台回复"性能工具"可以获取我整理的profiling脚本和压测配置。有问题评论区见，特别是关于selectinload和asyncpg的坑，我踩得比较深。