一、问题背景：一个"看起来很简单"的接口
事情是这样的。我们内部有个商品查询服务，最早是 Flask 写的，后来新接口迁到了 FastAPI，结果两套栈并存。出问题的是 /api/v1/products/search 这个接口：根据关键词搜索商品，返回商品基本信息 + 所属类目名 + 库存状态。
逻辑真的不复杂，但上线之后监控一直报警：
- P50：约 180ms
- P95：约 720ms
- P99：约 1200ms
- 超过 500ms 的请求占比 18%
用 wrk 在 4C8G 的测试机上压，300 并发直接开始超时，QPS 卡在 300 上不去。业务方天天来问，我只能硬着头皮去查。
这篇文章记录我整个排查和优化过程，包含 profiler 使用、SQL 优化、缓存设计，以及每一步的压测数据。
二、环境与版本
先把环境列清楚，避免大家踩版本坑：
| 组件 | 版本 |
|---|---|
| Python | 3.11.6 |
| FastAPI | 0.110.0 |
| Flask | 3.0.2 |
| Uvicorn | 0.27.1 |
| SQLAlchemy | 2.0.28 |
| asyncpg | 0.29.0 |
| psycopg2-binary | 2.9.9 |
| PostgreSQL | 15.5 |
| Redis | 7.2.4 |
| wrk | 4.2.0 |
部署方式：Uvicorn 4 worker，Nginx 反代。压测机是另一台同规格的 4C8G。
三、方案设计：先定位，再优化
我给自己定的顺序是：先 Profile 找到热点，再动 SQL 和缓存。不 Profile 直接改代码，是耍流氓。
整体思路：
- 用
cProfile抓单请求热点，用py-spy抓线上火焰图（不侵入） - 定位到 N+1 查询后，用 SQLAlchemy 的
selectinload+ 批量IN查询干掉 - 引入 Redis 两级缓存：本地 LRU（进程内）+ Redis（跨进程）
- FastAPI 侧把阻塞的 psycopg2 调用换成 asyncpg
- 每一步都压测对比，用数据说话
四、核心实现
4.1 用 cProfile + py-spy 定位瓶颈
先写个简单的压测脚本触发单请求：
# profile_single.py
import cProfile
import pstats
import requests
def hit():
# 模拟真实请求
for _ in range(20):
requests.get(
"http://127.0.0.1:8000/api/v1/products/search",
params={"q": "手机", "page": 1, "size": 20},
)
if __name__ == "__main__":
profiler = cProfile.Profile()
profiler.enable()
hit()
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(20)
跑完输出里最扎眼的两行：
ncalls tottime cumtime function
420 0.012 3.842 sqlalchemy/orm/query.py:2810(_load_for_path)
400 0.008 3.201 psycopg2/extras.py:1260(execute)
20 个请求打了 400 次 SQL，平均一个请求 20 条 SQL，典型的 N+1。
再用 py-spy 抓线上：
py-spy record -o profile.svg --pid $(pgrep -f uvicorn) --duration 30
火焰图显示 60% 时间卡在 psycopg2 的 execute 和网络等待上，剩下 20% 在 jsonable_encoder 上。
4.2 干掉 N+1：批量查询 + selectinload
原来的代码长这样（简化版）：
# 优化前
@app.get("/api/v1/products/search")
def search(q: str, page: int = 1, size: int = 20):
products = db.query(Product).filter(Product.name.ilike(f"%{q}%")) \
.offset((page - 1) * size).limit(size).all()
result = []
for p in products:
# 每条都查一次类目
category = db.query(Category).get(p.category_id)
# 每条都查一次库存
stock = db.query(Stock).filter(Stock.product_id == p.id).first()
result.append({
"id": p.id,
"name": p.name,
"category": category.name,
"stock": stock.quantity if stock else 0,
})
return result
一个请求 20 条商品，就要 1 + 20 + 20 = 41 条 SQL。改成批量：
# 优化后
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
@app.get("/api/v1/products/search")
async def search(q: str, page: int = 1, size: int = 20, db: AsyncSession = Depends(get_db)):
stmt = (
select(Product)
.where(Product.name.ilike(f"%{q}%"))
.options(selectinload(Product.category))
.offset((page - 1) * size)
.limit(size)
)
products = (await db.execute(stmt)).scalars().all()
# 一次性把库存查出来
pids = [p.id for p in products]
stock_map = {}
if pids:
rows = await db.execute(
select(Stock.product_id, Stock.quantity).where(Stock.product_id.in_(pids))
)
stock_map = {pid: qty for pid, qty in rows.all()}
return [
{
"id": p.id,
"name": p.name,
"category": p.category.name,
"stock": stock_map.get(p.id, 0),
}
for p in products
]
SQL 数量从 41 条降到 3 条（1 商品 + 1 类目 + 1 库存）。同时索引也要补上：
CREATE INDEX CONCURRENTLY idx_products_name_trgm
ON products USING gin (name gin_trgm_ops);
CREATE INDEX idx_stock_product_id ON stock (product_id);
4.3 两级缓存：本地 LRU + Redis
搜索接口的数据变化不频繁，特别适合缓存。我用的是本地 LRU + Redis 两级：
- L1：
cachetools.LRUCache，maxsize=2000，TTL 30s，进程内零网络开销 - L2：Redis，TTL 300s，跨进程共享
# cache.py
import json
import hashlib
from cachetools import TTLCache
from redis.asyncio import Redis
redis = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)
local_cache = TTLCache(maxsize=2000, ttl=30)
def make_key(prefix: str, **kwargs) -> str:
raw = json.dumps(kwargs, sort_keys=True, ensure_ascii=False)
return f"{prefix}:{hashlib.md5(raw.encode()).hexdigest()}"
async def get_cache(key: str):
# L1
if key in local_cache:
return local_cache[key]
# L2
val = await redis.get(key)
if val:
data = json.loads(val)
local_cache[key] = data
return data
return None
async def set_cache(key: str, value, ttl: int = 300):
local_cache[key] = value
await redis.set(key, json.dumps(value, ensure_ascii=False), ex=ttl)
接口里用法：
key = make_key("search", q=q, page=page, size=size)
cached = await get_cache(key)
if cached is not None:
return cached
# ... 查库逻辑 ...
result = [...]
await set_cache(key, result, ttl=300)
return result
踩坑提醒：别缓存 Depends 的 session 对象，也别缓存带时间戳的字段，否则本地缓存会返回过期数据。TTL 我设的是本地 30s / Redis 300s，业务能接受 30s 内数据不一致。
4.4 FastAPI 侧：把同步阻塞换成异步
Flask 那套默认是同步的，每个请求一个线程。FastAPI 虽然支持 async，但如果你的 DB driver 是 psycopg2，那还是阻塞的，会把 event loop 卡住。
换成 asyncpg：
# db.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
engine = create_async_engine(
"postgresql+asyncpg://user:pwd@127.0.0.1:5432/shop",
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=1800,
echo=False,
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
async with AsyncSessionLocal() as session:
yield session
Flask 侧我暂时没动，因为它在慢慢下线，只要保证不被新流量打爆就行。但 Flask 那边也加了 gunicorn --workers 8 --threads 4，比原来的 4 worker 好不少。
五、踩坑与优化
优化过程中踩了几个坑，记录一下：
-
selectinload和joinedload选错。一开始用joinedload，结果商品表 JOIN 类目表，行数膨胀导致去重开销大。selectinload发两条 SQL 但更稳，推荐。 -
Redis 连接没复用。最初每次请求
Redis()新建连接，QPS 一高就报Too many open files。改成模块级单例 + 连接池后正常。 -
本地缓存把内存吃满。
TTLCache的 maxsize 一开始设了 100000，一个进程吃了 2G 内存。调到 2000 后稳定在 200M 以内。 -
异步里混用同步代码。有一处日志用了同步写文件，QPS 高的时候直接卡 event loop。换成
logging.handlers.QueueHandler后解决。 -
压测机本身成瓶颈。一开始 wrk 和 uvicorn 跑在同一台机器上，怎么压都上不去。分开机器后数据才对。
六、效果数据
每一步压测都跑 60s，300 并发，wrk 命令：
wrk -t8 -c300 -d60s --latency \
"http://127.0.0.1:8000/api/v1/products/search?q=手机&page=1&size=20"
| 阶段 | QPS | P50 | P95 | P99 | 超时率 |
|---|---|---|---|---|---|
| 优化前 | 300 | 180ms | 720ms | 1200ms | 18% |
| 干掉 N+1 | 900 | 62ms | 180ms | 320ms | 2.1% |
| + Redis 缓存 | 1800 | 28ms | 78ms | 140ms | 0.3% |
| + asyncpg 异步 | 2400 | 19ms | 52ms | 85ms | 0.05% |
P99 从 1200ms 降到 85ms，QPS 从 300 到 2400，8 倍提升。缓存命中率稳定在 92% 左右，Redis 平均响应 0.6ms。
CPU 占用也从优化前的 4 核跑满，降到平均 35%。
七、总结
几点体会：
- 不要凭感觉优化。我一开始以为瓶颈在 JSON 序列化，结果 profile 一开，全在 SQL 上。
- N+1 是 API 性能第一杀手。一个接口 41 条 SQL，改完直接降 4 倍延迟。
- 缓存不是银弹，但用对了收益巨大。两级缓存要注意一致性和内存，TTL 别设太长。
- 异步不是免费的。FastAPI + asyncpg 提升明显，但如果代码里混了同步阻塞，反而更慢。
- 压测环境和生产环境要分开，否则你压的是压测机，不是服务。
代码我都贴出来了，可以直接拿去改。如果你也在做 API 性能优化，欢迎评论区交流你的瓶颈点在哪。
