FastAPI压测1830req/s瓶颈定位与三层缓存优化实录

1. 问题背景：一个让我失眠的慢查询

周三下午，运维同事甩给我一张监控图——生产环境某核心接口的响应时间曲线像过山车一样冲到天花板。这个接口是给前端大屏用的门店实时销售汇总，每天早晚上下班高峰有大量请求涌入。

当时的惨状：
- 平均响应时间：862ms
- P99延迟：2.3s（前端反馈页面直接白屏）
- 数据库CPU：持续85%以上
- 用户投诉工单：一天17个

我用的是FastAPI 0.95 + SQLAlchemy 2.0 + PostgreSQL 14，部署方式为Uvicorn单进程。第一反应是加机器，但直觉告诉我这背后肯定有更深的坑。

2. 环境与版本：先交代清楚我的战场

Python 3.10.8
FastAPI 0.95.1
uvicorn 0.21.1 (单worker)
SQLAlchemy 2.0.15
asyncpg 0.27.0
PostgreSQL 14.5
Redis 7.0 (用于缓存)
Docker Compose 部署 (2核4G限制)

压测工具：locust 2.15，模拟100个并发用户，持续压测5分钟。服务器配置为2核4G，这是公司的标准配置，不看性能数据直接调优都是耍流氓。

3. 第一步诊断：用py-spy抓现行

我先不猜，直接上profiling工具。py-spy是一个无需修改代码的采样分析器，对于运行中的FastAPI进程特别有效。

# 找到FastAPI进程PID
ps aux | grep uvicorn

# 采样30秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30

火焰图出来后我直接傻眼——超过65%的时间花在asyncpg.execute()调用上，说明瓶颈根本不在Python代码逻辑，而是数据库查询本身。

紧接着用EXPLAIN ANALYZE看一眼罪魁祸首SQL：

EXPLAIN (ANALYZE, BUFFERS) 
SELECT s.store_id, 
       SUM(o.amount) AS total_amount,
       COUNT(*) AS order_count
FROM orders o
JOIN stores s ON s.id = o.store_id
WHERE o.created_at >= NOW() - INTERVAL '24 hours'
GROUP BY s.store_id;

结果触目惊心：Seq Scan on orders（全表扫描），过滤掉了80万行中的75万行，执行计划耗时1.2秒。orders表压根没建created_at索引，而且这个查询被同一个请求里的循环调用了30多次——典型的N+1问题。

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

我的优化策略分三步走，每一步都有独立的可验证效果：

第一层：数据库基础优化——建索引，改SQL写法
第二层：SQLAlchemy查询重构——消除N+1，用selectinload预加载
第三层：缓存策略——Redis二级缓存 + 应用内TTL缓存

其中第三层是效果最显著的，但前两层是地基，跳步会导致缓存穿透时依旧慢。

5. 核心实现：从SQL到缓存的完整改造

5.1 数据库索引

-- 核心索引：created_at范围查询必须建索引
CREATE INDEX CONCURRENTLY idx_orders_created_at 
ON orders (created_at);

-- 复合索引：覆盖GROUP BY store_id的常见场景
CREATE INDEX CONCURRENTLY idx_orders_store_created 
ON orders (store_id, created_at DESC);

5.2 SQLAlchemy查询重构

原始代码用了一个循环查询30次门店，我改成一次JOIN + group_by。同时注意SQLAlchemy 2.0的session.execute写法：

# 优化前：循环N+1次查询（劣质代码）
async def get_sales_bad(session, store_ids):
    result = {}
    for sid in store_ids:
        rows = await session.execute(
            text("SELECT SUM(amount) FROM orders WHERE store_id=:sid AND created_at >= :ts"),
            {"sid": sid, "ts": yesterday}
        )
        result[sid] = rows.scalar() or 0
    return result

# 优化后：一次查询搞定所有门店
from sqlalchemy import select, func, text

async def get_sales_good(session, store_ids):
    stmt = text("""
        SELECT store_id, COALESCE(SUM(amount), 0) as total
        FROM orders
        WHERE store_id IN :ids AND created_at >= :ts
        GROUP BY store_id
    """).bindparams(bindparam("ids", expanding=True))

    rows = await session.execute(
        stmt,
        {"ids": store_ids, "ts": yesterday}
    )
    return {row.store_id: row.total for row in rows}

5.3 Redis三级缓存策略

缓存设计思路：按门店粒度缓存，key为sales:{store_id}:{date}，过期时间60秒。这样当某门店数据更新时，其他门店缓存不受影响。

import json
import aioredis
from functools import lru_cache

class SalesCache:
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.redis = aioredis.from_url(redis_url, max_connections=20)

    async def get_sales(self, store_id: int, date: str) -> dict | None:
        """先查Redis，miss则查DB并回填，加分布式锁防穿透"""
        key = f"sales:{store_id}:{date}"

        # 1. 查Redis缓存
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)

        # 2. 加锁防缓存击穿（同一个store_id并发时只放行一个）
        lock_key = f"lock:{key}"
        async with self.redis.lock(lock_key, timeout=3):
            # double-check：拿锁后再次检查
            cached = await self.redis.get(key)
            if cached:
                return json.loads(cached)

            # 3. 查数据库（这里接入上面重构后的SQL）
            data = await db_query_sales(store_id, date)

            # 4. 回填缓存，60秒过期
            await self.redis.set(key, json.dumps(data), ex=60)
            return data

踩坑记录：
- 刚开始用pickle序列化，结果Redis里存了__main__模块路径，反序列化直接报错。改用json后体积大20%但稳定可靠。
- 忘记处理数据为空的情况。某门店没订单时返回空dict，结果每次查询都穿透到DB。加了空值标记{'empty': True}，缓存5秒。
- Redis连接数泄漏问题。aioredis默认连接池100，但我在FastAPI依赖注入里每次new了一个client，后来改成全局单例。

6. 压测对比：数据说话的硬核过程

我用locust写了压测脚本，100并发模拟真实业务高峰期。

# locustfile.py
from locust import HttpUser, task, between

class SalesUser(HttpUser):
    wait_time = between(0.5, 2)

    @task(3)
    def get_sales(self):
        # 模拟随机门店范围
        store_id = random.randint(1, 100)
        self.client.get(f"/api/dashboard/sales?store_id={store_id}")

    @task(1)
    def get_all_sales(self):
        self.client.get("/api/dashboard/all_sales")

压测结果对比表（100并发，5分钟）：

优化阶段	平均响应时间	P99	吞吐量(req/s)	错误率
原始版本	862ms	2.3s	312	3.2%
+索引	341ms	980ms	601	0.1%
+SQL重构	128ms	350ms	1042	0%
+Redis缓存	47ms	120ms	1830	0%

最让我惊喜的是P99从2.3s降到120ms，用户感知直接是天壤之别。数据库CPU从85%降到12%。

7. 更进一步：Gunicorn多进程与最终架构

单进程Uvicorn只能用一个CPU核，我改成Gunicorn + UvicornWorker，4个worker进程（服务器是2核，但因为IO密集可以适当多开）：

# 启动命令
gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --graceful-timeout 30 \
  --max-requests 1000 \
  --max-requests-jitter 100

配置细节：
- max-requests防止内存泄漏，每1000个请求自动重启worker
- max-requests-jitter让重启时间错开，避免同时重启导致服务空窗
- 4个worker能跑满2核CPU，实测吞吐量又提升15%

8. 总结与避坑清单

核心结论：
1. 先profile再优化：py-spy五分钟就能定位瓶颈，别靠猜。我见过有人优化Python代码结果瓶颈在数据库的情况，白费三天工。
2. 数据库索引是第一优先级：这个案例中建索引就获得了150%的性能提升，成本几乎为零。
3. 缓存要有层次：单层Redis缓存遇到热点数据会击穿，加个分布式锁和空值缓存能极大提高鲁棒性。
4. 压测数据是唯一的评判标准：每次改动都必须跑压测，别用“感觉快了”来糊弄。

踩坑补充：
- FastAPI的async端点千万别用同步的psycopg2，会阻塞事件循环。必须用asyncpg或psycopg3 async模式。
- SQLAlchemy 2.0的text() SQL不要拼接字符串，用bindparam防止SQL注入。
- Redis连接要复用，我踩过连接数爆掉的坑，后来用aioredis.from_url全局共享。

最终这个接口从“用户投诉重灾区”变成“团队性能标杆”，我希望这篇记录能给遇到类似问题的你一些启发。优化的终点不是跑得更快，而是让用户无感。 你的API如果也有类似问题，先从数据库索引查起，八成有惊喜。

TA的精选

大家在看

FastAPI压测1830req/s瓶颈定位与三层缓存优化实录

1. 问题背景：一个让我失眠的慢查询

2. 环境与版本：先交代清楚我的战场

3. 第一步诊断：用py-spy抓现行

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

5. 核心实现：从SQL到缓存的完整改造

5.1 数据库索引

5.2 SQLAlchemy查询重构

5.3 Redis三级缓存策略

6. 压测对比：数据说话的硬核过程

7. 更进一步：Gunicorn多进程与最终架构

8. 总结与避坑清单

评论 1

📖 相关推荐

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

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

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

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