FastAPI慢查询根因追踪：Profiling与Redis缓存双层优化实录

一、问题背景：半夜报警，接口突然“卡死”

上周四凌晨，监控告警把我从床上叫醒——/api/v1/orders接口P99延迟飙升至2.3秒，错误率突破5%。这个接口是订单列表查询，平时P95稳定在800ms左右。

第一反应看日志，发现慢查询集中在数据库端：同一个SQL执行了上百次，每次耗时12-15ms。典型的N+1查询症状。更诡异的是，当我把数据库连接池调大后，延迟反而更高了——这说明瓶颈不在连接数，而在查询本身的效率。

二、环境与版本：技术栈全景

优化前，服务端技术栈如下：

Python: 3.10.12
FastAPI: 0.104.1
Uvicorn: 0.24.0 (workers=4, loop="uvloop")
SQLAlchemy: 2.0.23 (async模式)
PostgreSQL: 15.3 (连接池使用asyncpg)
Redis: 7.2.1 (用于缓存与分布式锁)

压测工具采用wrk，命令如下：

wrk -t8 -c200 -d60s --latency http://localhost:8000/api/v1/orders?user_id=1024

基线数据：请求量约1120 req/s，P95延迟820ms，数据库CPU占用率78%，服务端CPU占用率45%。

三、方案设计：三层漏斗排查法

我的优化策略分三步走，每一步都基于数据驱动，不靠猜：

Profiling定位热点：先用cProfile记录CPU调用栈，再用Py-Spy抓取运行时状态，确认是CPU密集还是IO密集。
数据库层优化：修正ORM查询策略，消除N+1，并利用EXPLAIN ANALYZE检查索引命中情况。
缓存层改造：引入两级缓存架构——本地内存缓存(LRU) + Redis分布式缓存，同时解决缓存击穿问题。

四、核心实现与代码演进

4.1 Profiling：直接证据链

先用py-spy dump抓取运行中的进程栈，连续采样5次：

py-spy dump --pid 12345 --duration 5

关键发现：sqlalchemy/loading.py中的_instance_processor占据62%的采样时间。这证实了N+1查询——每次迭代都触发独立的数据库往返。

接着用cProfile跑单请求：

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

client = TestClient(app)
profiler = cProfile.Profile()
profiler.enable()
response = client.get("/api/v1/orders?user_id=1024")
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(30)  # 打印前30行

输出关键行：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
  2500    0.012    0.000    1.642    0.001  /.../sqlalchemy/orm/loading.py:765(_instance_processor)

2500次调用，累计耗时1.6秒。这就是罪魁祸首。

4.2 数据库查询优化：从N+1到批量加载

原始代码使用了懒加载：

# 优化前：懒加载导致N+1
async def get_orders(user_id: int, db: AsyncSession):
    result = await db.execute(
        select(Order).where(Order.user_id == user_id)
    )
    orders = result.scalars().all()
    # 每次访问order.items都会触发新的SQL
    return [
        {"id": o.id, "items": [item.name for item in o.items]}
        for o in orders
    ]

修正为显式批量加载，使用selectinload一次关联查询：

# 优化后：使用selectinload批量加载
from sqlalchemy.orm import selectinload

async def get_orders_optimized(user_id: int, db: AsyncSession):
    stmt = (
        select(Order)
        .where(Order.user_id == user_id)
        .options(
            selectinload(Order.items)  # 生成IN查询，一次加载所有items
        )
    )
    result = await db.execute(stmt)
    orders = result.scalars().all()
    return [
        {"id": o.id, "items": [item.name for item in o.items]}
        for o in orders
    ]

效果验证：数据库查询次数从原来的1 + N次降为1 + 1次。用EXPLAIN确认走了user_id_idx索引，且没有额外的SEQ SCAN。

4.3 缓存策略：多级缓存与锁防击穿

SQL优化后P95延迟降至280ms，但仍不够。我发现这个接口的读多写少特性——同一用户的订单数据在5分钟内变化率低于0.1%。于是引入缓存：

# 两级缓存：本地LRU + Redis分布式缓存
from functools import lru_cache
import aioredis
import json

class OrderCache:
    def __init__(self, redis_url: str):
        self.redis = aioredis.from_url(redis_url)
        self.local_ttl = 30      # 本地缓存30秒
        self.redis_ttl = 300     # Redis缓存5分钟

    @lru_cache(maxsize=256)
    def _local_cache(self, key: str):
        return None  # 占位符，实际用字典存储

    async def get_or_set(self, user_id: int, db_loader):
        # 1. 查询本地缓存
        cache_key = f"orders:{user_id}"
        local_data = self._local_cache(cache_key)
        if local_data:
            return local_data

        # 2. 查询Redis
        redis_data = await self.redis.get(cache_key)
        if redis_data:
            # 回填本地缓存
            data = json.loads(redis_data)
            self._local_cache(cache_key, data)
            return data

        # 3. 缓存未命中 -> 查数据库，加分布式锁防击穿
        async with self.redis.lock(f"lock:{cache_key}", timeout=5):
            # 双重检查，防止锁等待后重复查询
            redis_data = await self.redis.get(cache_key)
            if redis_data:
                return json.loads(redis_data)
            data = await db_loader()
            await self.redis.set(cache_key, json.dumps(data), ex=self.redis_ttl)
            self._local_cache(cache_key, data)  # 注意设置ttl需要自定义
            return data

踩坑记录：lru_cache不支持单独设置TTL且不可控，我改用cachetools.TTLCache替代：

from cachetools import TTLCache

self.local_cache = TTLCache(maxsize=512, ttl=30)

同时，为了压测时避免缓存干扰，我在测试脚本中加入了Cache-Control头模拟真实用户。

五、踩坑与优化：压测中的意外发现

坑1：Uvicorn workers数与连接池不匹配
最初配置workers=8，但每个worker独立维护数据库连接池，导致连接数爆炸。最终固定为workers=4，并设置max_overflow=20。

坑2：Redis锁的续期问题
当数据库查询超过5秒（比如订单量大时），锁自动过期导致并发查询穿透。解决方案是租约续期，使用Redlock算法。但考虑到业务场景，我直接优化了SQL，确保单次查询低于200ms。

坑3：压测数据污染缓存
wrk产生的连续请求会让缓存命中率接近100%，掩盖真实问题。我在压测请求头中加入随机参数：

wrk -t8 -c200 -d60s --latency -H "Cache-Control: no-cache" http://localhost:8000/api/v1/orders?user_id=1024&_=random

六、效果数据：从820ms到97ms的完整记录

优化后的最终压测结果（使用同一台wrk机器）：

指标	优化前	优化后	提升幅度
P95延迟	820ms	97ms	88.2%↓
吞吐量	1120 req/s	5150 req/s	359%↑
数据库QPS	2100	340	83.8%↓
服务端CPU	45%	23%	49%↓
错误率	5.2%	0.01%	99.8%↓

性能曲线：P99从2.3s降至150ms，完全满足SLA要求。

七、经验总结与教训

ORM不背锅，但懒加载必须死：SQLAlchemy的N+1是常见陷阱，用selectinload或joinedload做批量加载。
Profiling先行：不要凭感觉优化。py-spy dump 30秒就能定位热点，比瞎猜高效十倍。
缓存要考虑击穿：高并发下必须用分布式锁保护，否则缓存失效瞬间会打垮数据库。
压测要模拟真实场景：关闭浏览器缓存，加随机参数，否则结果会过于乐观。

最后提醒：优化是无限游戏，这版缓存架构已经稳定运行两周，目前正在尝试用Cython编译热点函数，期待下一轮数据。

附录：完整优化后的关键依赖版本

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy[asyncio]==2.0.23
asyncpg==0.29.0
redis==5.0.1
cachetools==5.3.2
py-spy==0.3.14

如果你也遇到类似性能问题，建议先跑一遍py-spy dump，再考虑加缓存。不要一上来就上Redis，先消除N+1，再谈缓存。

TA的精选

大家在看

FastAPI慢查询根因追踪：Profiling与Redis缓存双层优化实录

一、问题背景：半夜报警，接口突然“卡死”

二、环境与版本：技术栈全景

三、方案设计：三层漏斗排查法

四、核心实现与代码演进

4.1 Profiling：直接证据链

4.2 数据库查询优化：从N+1到批量加载

4.3 缓存策略：多级缓存与锁防击穿

五、踩坑与优化：压测中的意外发现

六、效果数据：从820ms到97ms的完整记录

七、经验总结与教训

评论 1

📖 相关推荐

Git分支策略与CI/CD流水线：团队协作的版本控制实践

Git工作流实践：从分支策略到CI/CD集成的团队协作方案

FastAPI异步重构与Redis缓存：API响应从2.3秒降至95ms

FastAPI性能调优：Profiling、SQL查询与缓存策略压测实录