FastAPI压测1200qps到4800qps：Profiling与缓存三层优化实录

1. 问题背景：凌晨两点被报警电话叫醒

事情是这样的，我们有个内部数据看板服务，FastAPI写的，部署在K8s上（3个Pod，每个2核4G）。平时白天峰值也就800qps左右，结果上周二晚上数据仓库跑批任务异常，导致下游服务疯狂重试，直接把这个API打到1200qps。

然后就是熟悉的剧情：CPU直接打满，P99延迟从80ms飙到2.3s，报警电话打到我手机上。更烦的是，我看了下监控，发现CPU高得离谱，但数据库连接池和慢查询日志都没啥异常——直觉告诉我，问题出在应用层，而不是数据库层。

2. 环境与版本：先交代清楚基线

• Python 3.11.4 + FastAPI 0.104.1 + Uvicorn 0.24.0（workers=2，每个worker配了gthread模式，threads=16）
• SQLAlchemy 2.0.21 + asyncpg 0.28.0（异步连接池，pool_size=20, max_overflow=10）
• Redis 7.0.12（单节点，maxmemory 2GB，allkeys-lru淘汰策略）
• 部署：Docker镜像基于python:3.11-slim，K8s资源限制 CPU=2.0，内存=4Gi
• 压测工具：wrk 4.2.0，单机跑在另一台8核16G的机器上（避免压测机本身成为瓶颈）

压测命令统一用：wrk -t8 -c200 -d60s --latency http://api.example.com/v1/orders/summary?date=2024-03-15

3. 第一步：Profiling先行，别瞎猜

我的原则是：性能优化第一件事永远是找证据，不是拍脑袋改代码。这里用了两个工具：

3.1 py-spy抓CPU热点

py-spy是个好工具，不用改代码就能拿到Python进程的调用栈和CPU占用分布。在容器里执行：

# 在宿主机上对容器内PID抓取CPU火焰图数据
py-spy record --pid $(pgrep -f uvicorn | head -1) --duration 30 --output flamegraph.svg

看火焰图，顶部最宽的几个函数是：
- sqlalchemy.orm.session.Session.execute 占了 38% 的CPU时间
- pydantic.BaseModel.dict 和 fastapi.encoders.jsonable_encoder 合计 22%
- asyncio.tasks 的调度开销 12%

3.2 数据库慢查询日志

打开PostgreSQL的auto_explain模块，设置阈值100ms：

LOAD 'auto_explain';
SET auto_explain.log_min_duration = 100;
SET auto_explain.log_analyze = true;

结果发现有个查询平均执行 80ms，但被调用了 N+1次（订单列表每条记录查一次用户信息）。单次80ms不慢，但200条订单就是16秒的累积等待。

结论：瓶颈不在数据库本身，而在应用层的查询次数过多和序列化开销过大。数据库连接池反而没打满，说明不是连接泄露之类的问题。

4. 第二步：数据库查询优化——消灭N+1

原始代码（简化版）：

# 问题代码：循环内查询，N+1噩梦
async def get_orders_summary(date: str):
    orders = await db.execute(
        select(Order).where(Order.created_at >= date)
    )
    result = []
    for order in orders.scalars().all():
        # 每条订单查一次用户！这就是N+1
        user = await db.execute(
            select(User).where(User.id == order.user_id)
        )
        result.append({
            "order_id": order.id,
            "user_name": user.scalar().name,
            "amount": order.amount
        })
    return result

优化方案：用selectinload一次性加载关联实体，避免循环查询。

# 优化代码：预加载关联，消灭N+1
from sqlalchemy.orm import selectinload

async def get_orders_summary(date: str):
    orders = await db.execute(
        select(Order)
        .options(selectinload(Order.user))  # 关键：预加载user
        .where(Order.created_at >= date)
    )
    result = []
    for order in orders.scalars().all():
        result.append({
            "order_id": order.id,
            "user_name": order.user.name,  # 不再触发查询
            "amount": order.amount
        })
    return result

效果数据：数据库查询次数从1 + N次降为2次（一条订单查询+一条用户批量查询）。这一步单独压测，qps从1200提升到2300，P99延迟从2.3s降到400ms。

踩坑提示：selectinload不是万能的——如果你只查20个字段里的2个，它会多查很多无关列。可以考虑用load_only或写原生SQL只查需要的列。但在这个场景下，全字段加载的额外开销可以接受。

5. 第三步：序列化优化——Pydantic v2还是原生dict？

火焰图显示pydantic序列化占了22%CPU。我试了两个方案：

方案A：用Pydantic v2（基于Rust）替换Pydantic v1

FastAPI 0.104默认用的Pydantic v2（2.4.2），但我看到代码里显式用了from pydantic import BaseModel，检查发现是v1的写法（orm_mode）。改用v2的model_config写法后，序列化速度提升约 3倍。具体改动：

# Before (Pydantic v1 style)
class OrderSummary(BaseModel):
    order_id: int
    user_name: str
    amount: float
    class Config:
        orm_mode = True

# After (Pydantic v2 style)
from pydantic import BaseModel, ConfigDict

class OrderSummary(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # v2写法
    order_id: int
    user_name: str
    amount: float

方案B：绕过Pydantic，直接返回dict

如果这个API只给前端用，不需要严格的数据校验，可以直接返回dict，省掉整个序列化层。但这样会丢失类型检查和文档生成能力，我最终没全量采用，只在最内层的列表项用了dict：

# 最终实现：混合策略，外快内dict
async def get_orders_summary(date: str):
    orders = await db.execute(
        select(Order).options(selectinload(Order.user)).where(Order.created_at >= date)
    )
    # 直接构建dict列表，不经过Pydantic
    return [
        {
            "order_id": o.id,
            "user_name": o.user.name,
            "amount": round(float(o.amount), 2)  # 避免Decimal序列化开销
        }
        for o in orders.scalars().all()
    ]

效果数据：这一步单独压测，qps从2300提升到3100。P99延迟从400ms降到250ms。

踩坑提示：Decimal类型序列化很慢。我们金额字段用的Numeric(10,2)，返回前手动round(float(...), 2)，虽然浮点数有精度损失风险，但在这个报表场景可接受。如果你在意精度，可以全局注册一个自定义JSON encoder。

6. 第四步：缓存策略——Redis缓存热点数据

数据库查询优化和序列化优化做完，qps到了3100，但还不够，因为下游重试流量还在涨。继续看火焰图，发现asyncpg的IO等待占了15%——数据库连接池的20个连接已经不够用了。这时候上缓存。

缓存策略设计：
- 缓存key：order_summary:{date}，TTL设60秒（业务上允许1分钟延迟）
- 缓存穿透：如果查询结果为空，也缓存空列表，TTL设为5秒，防止恶意请求打穿
- 缓存雪崩：TTL加随机抖动（±10秒），防止同一时间过期
- 缓存更新：不做主动失效，靠TTL自然过期。因为数据是T+1的报表，不需要实时性

import json
import random
import redis.asyncio as aioredis

redis_client = aioredis.from_url(
    "redis://redis-service:6379/0",
    decode_responses=True,
    max_connections=50
)

async def get_orders_summary_cached(date: str):
    cache_key = f"order_summary:{date}"

    # 先查缓存
    cached = await redis_client.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    # 缓存未命中，查数据库（注意：这里是优化后的查询）
    orders = await db.execute(
        select(Order).options(selectinload(Order.user)).where(Order.created_at >= date)
    )
    result = [
        {
            "order_id": o.id,
            "user_name": o.user.name,
            "amount": round(float(o.amount), 2)
        }
        for o in orders.scalars().all()
    ]

    # 写入缓存，TTL加随机抖动防止雪崩
    ttl = 60 + random.randint(-10, 10)
    if not result:
        ttl = 5  # 空结果短TTL，防穿透

    await redis_client.set(cache_key, json.dumps(result), ex=ttl)
    return result

效果数据：这一步单独压测，qps从3100提升到4800。P99延迟从250ms降到120ms。数据库连接池的使用率从95%降到40%，基本消除了数据库压力。

踩坑提示：
1. 序列化格式：我用json.dumps但没指定ensure_ascii=False，导致中文被转成\uXXXX，体积大30%。后来改成json.dumps(result, ensure_ascii=False)，带宽省了不少。
2. Redis连接池：一开始用redis.Redis（同步客户端）在async函数里调用，直接阻塞事件循环，qps反而降到800。必须用redis.asyncio（异步客户端）。
3. 缓存Key设计要防热点：如果所有请求都打同一个date（比如今天），那只有第一个请求会查库，没问题。但如果业务有大量不同日期，缓存命中率会降低，需要更细粒度的key设计。

7. 最终效果与总结

最终全量优化后峰值能到5000+qps（压测机8线程200连接上限了，再高需要加压测机并发数），P99稳定在90ms左右，CPU从打满降到30%，还有余量应对突发流量。

核心心得：
1. 先profiling再动手，py-spy + 慢查询日志是最快定位问题的方式。别猜，猜必错。
2. 数据库优化优先于缓存，N+1查询是万恶之源，先消灭它再谈缓存。缓存是加速器，不是救火队员。
3. 序列化开销被严重低估，尤其在高qps场景，Pydantic v1的序列化性能是硬伤。能用dict就别用ORM对象，能用v2就别用v1。
4. 缓存要考虑穿透、雪崩、热点，不能只写set/get就完事。TTL抖动、空结果缓存、连接池复用都是必修课。

最后说一句，这个API在压测后上线，第二天又遇到一次数据仓库重跑，这次qps冲到1800，P99只到180ms，稳如老狗。调优这事，工具和方法论比盲人摸象式的直觉重要得多。