FastAPI接口延迟飙到2.8秒，我用cProfile和Redis把P95砍到180ms

一、问题背景：一个聚合接口，P95 2.8秒

事情是这样的，上周三下午，业务方的产品经理直接在IM上甩了一个截图——我们内部数据平台的/api/v1/dashboard/summary接口，页面加载转圈转了整整三秒。截图下面配了一句：“这玩意儿还能用吗？”

我赶紧查了一下监控面板，果然，这个接口的P95延迟已经飙到2.8秒，P99更是惨不忍睹的4.1秒。这个接口的作用是聚合用户维度、订单维度、商品维度等7张表的数据，返回一个综合看板。之前数据量小的时候没什么感觉，但这个月业务量涨了大概3倍，问题就爆发了。

说实话，这个接口我自己写的，当时图省事，直接用SQLAlchemy ORM一顿查，也没想太多。现在想想，活该被骂。

二、环境与版本

先交代一下环境，方便大家复现：

• Python 3.10.12
• FastAPI 0.104.1 + Uvicorn 0.24.0（worker=2, 异步模式）
• SQLAlchemy 2.0.23（用asyncpg驱动连PostgreSQL 14.5）
• Redis 7.0.12（用redis-py 5.0.1，使用连接池）
• 部署：Docker容器，4核8G限制
• 压测工具：wrk 4.2.0

硬件条件一般，但关键点在于，这个接口的QPS其实不高，峰值也就100左右，问题纯粹是单次请求太慢。

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

很多同学遇到性能问题第一反应是“加缓存”、“加索引”，但我的习惯是先拿到证据。这里我用两个工具：

3.1 cProfile：快速看函数级耗时

因为接口是异步的，cProfile在线程级别不好使，但我可以在接口入口包一层同步包装器，用cProfile跑一次完整请求：

import cProfile
import pstats
import io

async def profile_wrapper():
    # 模拟一次真实请求
    from app.api.dashboard import generate_summary
    result = await generate_summary(user_id=12345, date_range="2024-11-01:2024-11-30")
    return result

profiler = cProfile.Profile()
profiler.enable()
asyncio.run(profile_wrapper())
profiler.disable()

stream = io.StringIO()
stats = pstats.Stats(profiler, stream=stream).sort_stats('cumulative')
stats.print_stats(40)
print(stream.getvalue())

输出结果中最扎眼的一行是：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
   127    0.002    0.000    1.945    0.015  sqlalchemy/orm/loading.py:670(_load_scalar_from_orm)

_load_scalar_from_orm被调用了127次，累计耗时1.945秒。这基本就是N+1查询的实锤了——我在循环里对每个用户查了一次订单数，对每个订单又查了一次商品明细。

3.2 py-spy：看实时调用栈

cProfile能给出函数级数据，但有时候我想看“当前卡在哪个IO上”，这时候用py-spy直接抓运行中的进程：

# 在容器内执行
py-spy dump --pid 12345

抓出来的栈显示，大量协程阻塞在asyncpg的_read_message上，说明数据库连接已经饱和，很多查询在排队等待连接。这就解释了为什么延迟是“非线性上涨”——QPS稍微一高，连接池耗尽，全部请求开始排队。

结论：瓶颈主要有三个：
1. N+1查询导致SQL执行次数过多（127次查询）
2. JSON序列化太慢（默认json库处理大字典时CPU占用高）
3. 没有缓存，每次请求都全量计算聚合结果

四、方案设计：三管齐下

针对上面三个瓶颈，我的优化策略如下：

另外还顺手做了个调整：把asyncpg连接池的max_size从5调到了20，因为之前连接池太小导致排队。

五、核心实现：代码与踩坑

5.1 优化查询：selectinload

原来的代码大概是这样的（示意）：

# 优化前：每次循环都查询一次
users = await session.execute(select(User).where(User.tenant_id == tenant_id))
for user in users.scalars():
    orders = await session.execute(select(Order).where(Order.user_id == user.id))
    ...

优化后，一次性用selectinload把关联对象全捞出来：

# 优化后：一次性加载所有关联
from sqlalchemy.orm import selectinload

async def get_summary_data(tenant_id: int, date_range: tuple):
    stmt = (
        select(User)
        .options(
            selectinload(User.orders).selectinload(Order.items),
            selectinload(User.profile),
        )
        .where(User.tenant_id == tenant_id)
        .limit(500)
    )
    result = await session.execute(stmt)
    users = result.scalars().unique().all()
    # 后续直接在内存中聚合，不再触发数据库查询

踩坑1：selectinload如果嵌套层级太深（比如上面的User.orders.items），生成的SQL会有多个IN子句，数据量大时反而变慢。我的解决方法是只加载两层，第三层（商品详情）改为单独一次查询后用字典映射。

踩坑2：SQLAlchemy 2.0的scalars().unique()必须加上，否则因为selectinload会产生重复行，结果集长度会翻倍，导致聚合逻辑出错。这个坑我调了半小时才定位到。

5.2 缓存策略：Redis + 手动防穿透

缓存设计思路很简单：

• Key = summary:{tenant_id}:{date_range}
• Value = JSON字符串（orjson序列化）
• TTL = 60秒

但这里有个大坑：缓存穿透。如果某个租户的数据量特别大，第一次请求还没算完，第二个请求又来了，两个请求都在跑相同的聚合SQL，数据库直接被打满。

解决办法是用Redis的SETNX（SET if Not Exists）做互斥锁：

import orjson
import redis.asyncio as aioredis

redis_client = aioredis.from_url("redis://localhost:6379/0", max_connections=10)

async def get_cached_summary(tenant_id: int, date_range: str):
    cache_key = f"summary:{tenant_id}:{date_range}"

    # 先尝试读缓存
    cached = await redis_client.get(cache_key)
    if cached:
        return orjson.loads(cached)

    # 尝试获取锁，防止缓存穿透
    lock_key = f"lock:{cache_key}"
    got_lock = await redis_client.set(lock_key, "1", nx=True, ex=10)
    if not got_lock:
        # 没拿到锁，说明有其他请求在计算，短暂等待后重试
        await asyncio.sleep(0.1)
        return await get_cached_summary(tenant_id, date_range)

    try:
        # 计算聚合结果
        data = await compute_summary_from_db(tenant_id, date_range)
        # 写入缓存
        await redis_client.set(cache_key, orjson.dumps(data), ex=60)
        return data
    finally:
        # 释放锁
        await redis_client.delete(lock_key)

踩坑3：Redis连接池默认max_connections=10，在异步环境下如果不显式设置，多个协程并发时会抛ConnectionPoolError。所以一定要在from_url里指定max_connections。

踩坑4：锁的过期时间设为10秒，但如果计算逻辑超过10秒，锁会被自动释放，导致两个请求同时计算。我的解决方法是把计算逻辑拆分成更小的批次，保证单次计算不超过5秒（500个用户限制），同时锁的过期时间设为30秒作为兜底。

5.3 序列化优化：orjson

FastAPI默认使用json.dumps，对于复杂嵌套字典，速度确实不行。切换方法很简单，在FastAPI初始化时指定：

import orjson
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

注意：如果某些接口返回的是Pydantic模型，需要确保模型能正确序列化。ORJSONResponse对datetime等类型的处理方式略有不同，我这边统一在Pydantic模型中做了字段格式转换。

六、压测数据：wrk实测结果

优化完成后，我用wrk跑了一轮压测，参数如下：

wrk -t4 -c100 -d30s --latency http://localhost:8000/api/v1/dashboard/summary?tenant_id=1001&date_range=2024-11-01:2024-11-30

顺便说一句，压测前我把Redis里的缓存主动清空了，确保测的是冷启动场景。

另外看一个更直观的wrk输出片段（优化后）：

  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   112.22ms   45.67ms 580.00ms   85.00%
    Req/Sec   223.44     38.12   380.00     76.00%

注意：由于是冷启动，每次请求都要查数据库，这个数字还有进一步提升空间。如果缓存命中（TTL内第二次请求），延迟可以稳定在40ms以下。

七、总结与一些感想

这次调优的核心思路其实就是先定位后优化，没有一上来就盲改。总结一下几个关键点：

1. 性能调优的第一步永远是profiling。cProfile和py-spy配合使用，一个看函数级耗时，一个看实时阻塞点。没有数据支撑的优化都是瞎猜。
2. N+1查询是Python后端性能的头号杀手。SQLAlchemy的selectinload非常好用，但要注意层级深度，建议最多两层，再多就要考虑手工分批查询。
3. 缓存不是银弹。缓存穿透、缓存击穿、缓存雪崩这三个问题在并发场景下都会出现，我的方案用SETNX加锁解决了穿透，用TTL随机化（60秒±10秒）缓解了雪崩。
4. 别忽略序列化开销。在高QPS下，JSON序列化的CPU占用会让你怀疑人生。orjson是纯C扩展，性能碾压标准库，换起来成本极低。

最后说句实话，这个接口之前写的确实烂，但经过这轮调优，我现在对自己的代码有信心多了。下次再遇到性能问题，记住：先侧录，再动手，别让感觉指挥你的键盘。

参考版本列表：
- Python 3.10.12
- FastAPI 0.104.1
- SQLAlchemy 2.0.23
- asyncpg 0.29.0
- redis-py 5.0.1
- orjson 3.9.7
- wrk 4.2.0