FastAPI+Flask接口从1.2s降到80ms：一次API性能瓶颈定位与调优记录

一、问题背景

事情是这样的。我们有个商品详情接口 /api/v1/product/{id}，早期用Flask写的，一直跑得还行。最近运营搞活动，流量涨了三倍，接口直接崩了。

监控上看到的现象：

P99 响应时间：1200ms（正常应该在100ms以内）
平均响应时间：380ms
QPS 到 200 左右开始出现 502
CPU 打满，但数据库连接池没打满

这就很诡异。CPU 高但DB连接没满，说明瓶颈要么在应用层的计算，要么在序列化，要么就是查询次数太多但每次很快。先别猜，上工具。

二、环境与版本

先把环境交代清楚，不然复现不了：

Python 3.10.13
Flask 2.3.3 + Gunicorn 21.2.0（4 workers，sync worker）
FastAPI 0.109.0 + Uvicorn 0.27.0
SQLAlchemy 2.0.25
PostgreSQL 14.10
Redis 7.2.3
机器：4C8G，SSD

三、定位瓶颈：py-spy + 慢查询日志

3.1 用 py-spy 抓火焰图

线上不敢随便 attach debugger，py-spy 是最稳的选择，几乎零开销：

pip install py-spy==0.3.14

# 找到 gunicorn worker 的 pid
ps aux | grep gunicorn

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

打开 profile.svg 一看，超过 60% 的采样时间堆在 SQLAlchemy 的 session.execute 上，而且调用栈里反复出现同一个函数 get_product_detail。再细看，这个函数里有个循环：

# 老代码，问题所在
def get_product_detail(product_id):
    product = Product.query.get(product_id)
    # N+1 元凶：循环里查 SKU，每个 SKU 又查库存
    for sku in product.skus:
        sku.stock = Inventory.query.filter_by(sku_id=sku.id).first().stock
    return product.to_dict()

一个商品平均 8 个 SKU，等于一次请求打 1 + 8 + 8 = 17 次 SQL。QPS 200 的时候，数据库每秒要扛 3400 次查询，PG 直接跪。

3.2 用 PG 慢查询日志确认

-- postgresql.conf
log_min_duration_statement = 100  -- 记录超过100ms的

慢日志里清一色是 SELECT * FROM inventory WHERE sku_id = $1，单次 2-5ms，但次数太密。

结论：瓶颈是 N+1 查询 + 无缓存，应用层 CPU 高是因为 ORM 反序列化开销大。

四、方案设计

方向定了：

消除 N+1：用 JOIN 一次把 SKU 和库存捞出来
加缓存：商品详情读多写少，Redis 缓存 60 秒，命中率能到 95%+
换框架：Flask 的 sync worker 在高并发下切换开销大，换成 FastAPI + Uvicorn（async），配合 asyncpg
加索引：inventory.sku_id 加索引

整体架构：

Client → Nginx → Uvicorn(FastAPI) → Redis(命中直接返回)
                                  ↓ miss
                              PostgreSQL (JOIN 一次查询)

五、核心实现

5.1 优化后的查询：一次 JOIN 搞定

# models.py
from sqlalchemy import select, join
from sqlalchemy.ext.asyncio import AsyncSession

async def fetch_product_detail(session: AsyncSession, product_id: int):
    stmt = (
        select(Product, Sku, Inventory)
        .join(Sku, Sku.product_id == Product.id)
        .join(Inventory, Inventory.sku_id == Sku.id)
        .where(Product.id == product_id)
    )
    result = await session.execute(stmt)
    rows = result.all()
    if not rows:
        return None
    product = rows[0][0]
    product.skus = [
        {"id": s.id, "name": s.name, "stock": inv.stock}
        for _, s, inv in rows
    ]
    return product

17 次 SQL → 1 次 SQL。

5.2 FastAPI 接口 + Redis 缓存

# main.py
import json
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis

app = FastAPI()
redis = Redis(host="127.0.0.1", port=6379, decode_responses=True)

CACHE_KEY = "product:detail:{pid}"
CACHE_TTL = 60  # 秒

@app.get("/api/v1/product/{product_id}")
async def get_product(product_id: int):
    key = CACHE_KEY.format(pid=product_id)
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)

    async with async_session() as session:
        product = await fetch_product_detail(session, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="product not found")

    data = {
        "id": product.id,
        "name": product.name,
        "skus": product.skus,
    }
    await redis.setex(key, CACHE_TTL, json.dumps(data))
    return data

启动命令：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop

注意 --loop uvloop，事件循环性能能提升 20% 左右。

5.3 索引

CREATE INDEX CONCURRENTLY idx_inventory_sku_id ON inventory(sku_id);
CREATE INDEX CONCURRENTLY idx_sku_product_id ON sku(product_id);

CONCURRENTLY 是为了不锁表，线上加索引必备。

六、踩坑与优化

坑1：缓存雪崩。 一开始 TTL 全设 60 秒，结果整点批量失效，DB 瞬间被打爆。后来在 TTL 上加了 ±10 秒的随机抖动：

import random
ttl = CACHE_TTL + random.randint(-10, 10)
await redis.setex(key, ttl, json.dumps(data))

坑2：asyncpg + SQLAlchemy 的坑。 AsyncSession 用完之后一定要 await session.close()，否则连接池很快耗尽。我用了 async with 上下文管理器才行。

坑3：序列化开销。 一开始用 Pydantic 模型返回，response_model 在数据量大时序列化开销明显。商品详情字段不多，直接返回 dict 反而更快，实测 P99 差了 15ms 左右。当然如果字段多、要做校验，还是老实上 Pydantic。

坑4：Flask 迁移成本。 其实不是所有接口都值得迁 FastAPI。我们只迁了 QPS 最高的 3 个接口，其他还跑 Flask，两套框架用 Nginx 按路径分流。

七、效果数据

用 wrk 压测，同一台机器，60 秒，200 并发连接：

wrk -t8 -c200 -d60s --latency http://127.0.0.1:8000/api/v1/product/1001

指标	优化前(Flask)	优化后(FastAPI+Redis)
QPS	180	1900
P50	320ms	12ms
P99	1200ms	80ms
错误率	4.7%	0%
CPU 使用率	98%	45%

Redis 缓存命中率 96.3%（redis-cli info stats 里看的 keyspace_hits / (hits+misses)）。

八、总结

这次调优核心就三件事：先用 py-spy 找到真瓶颈，别瞎猜；消 N+1 是收益最高的；缓存是第二收益。框架迁移其实是锦上添花，如果只做前两步，QPS 也能到 800 左右。

几点经验：

profiling 一定要上生产环境采样，本地复现不出真实流量特征
N+1 是 API 性能第一杀手，ORM 用着爽，埋雷也深
缓存 TTL 加随机抖动，这条几乎是肌肉记忆了
别为了性能盲目上 async，IO 密集才有效，CPU 密集的接口换 async 反而更慢

后续打算把商品列表接口也照这个思路过一遍，那个接口的 N+1 更夸张，一个列表 20 个商品，能打出 300+ SQL。有进展再写一篇。

大家在看