上周五下午，运维同学在群里丢了一张压测报告截图，说新上线的商品列表接口有点问题。我看了眼数据：P99 2.8s，TPS 180。当时心里咯噔一下——这接口是给首页用的，晚上大促就要开始了。
说实话，这个接口逻辑并不复杂：查商品表，关联分类、品牌、SKU，然后分页返回。之前开发时本地跑着挺快，一上压测就现原形。今天把这三天排查优化的过程完整记录下来，希望能帮到遇到类似问题的朋友。
1. 环境与版本说明
先交代一下运行环境，方便大家对照：
Python 3.10.12
FastAPI 0.95.1
uvicorn 0.21.1 (workers=4)
SQLAlchemy 2.0.15
PostgreSQL 14.2
Redis 7.0 (单机)
压测工具：wrk 4.2.0 (线程4，连接100，时长30s)
机器：8C16G 云主机，PG和Redis同机部署
压测命令统一使用：wrk -t4 -c100 -d30s http://localhost:8000/api/v1/products?page=1&size=20
2. 第一轮排查：cProfile定位CPU热点
先用cProfile跑了一次单请求分析，采样100次取中位数：
import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient
client = TestClient(app)
profiler = cProfile.Profile()
profiler.enable()
for _ in range(100):
client.get("/api/v1/products?page=1&size=20")
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumtime").print_stats(20)
输出关键行（截取前5条）：
ncalls tottime percall cumtime percall filename:lineno(function)
100 0.182 0.002 8.734 0.087 /app/models.py:86(get_products)
100 0.001 0.000 7.912 0.079 /app/schemas.py:44(product_to_dict)
100 0.003 0.000 6.893 0.069 /app/services.py:31(ProductService.list)
明显了，product_to_dict 和 get_products 占了大头。product_to_dict 里面做了什么？仔细看代码，原来每个商品要拼SKU列表和分类路径，分别查了两次数据库。20个商品，就是20*(1+2) = 60次查询。
这就是教科书级的N+1问题。
3. 第一轮优化：SQLAlchemy selectinload
先看原始查询代码（简化版）：
# 优化前
products = db.query(Product).offset(offset).limit(size).all()
result = []
for p in products:
skus = db.query(SKU).filter(SKU.product_id == p.id).all()
category = db.query(Category).filter(Category.id == p.category_id).first()
result.append({
"id": p.id,
"name": p.name,
"skus": [s.to_dict() for s in skus],
"category": category.name
})
改成SQLAlchemy 2.0的selectinload：
# 优化后
from sqlalchemy.orm import selectinload
stmt = (
select(Product)
.options(
selectinload(Product.skus),
selectinload(Product.category),
)
.offset(offset)
.limit(size)
)
products = db.execute(stmt).scalars().all()
result = [p.to_dict() for p in products]
同时把Product.to_dict()写成了方法，内部直接遍历已加载的关系，不再触库。
压测一下（wrk 30s）：
| 指标 | 优化前 | 优化后 | 变化 |
|---|---|---|---|
| TPS | 180 | 640 | +255% |
| P50 | 480ms | 150ms | -69% |
| P99 | 2800ms | 620ms | -78% |
数据库查询次数从每次请求60次降到了3次（主表+SKU批量IN+分类批量IN）。效果显著，但说实话还不够——P99还是600多毫秒，对首页来说不能忍。
4. 第二轮优化：Redis缓存策略
单纯靠SQL优化已经到瓶颈了，再往下就是加缓存。考虑到商品数据变更频率低（管理员后台修改），我们决定用Redis做两级缓存：列表缓存 + 单品缓存。
缓存策略设计：
- 列表缓存：key为
products:list:{page}:{size}，缓存5分钟 - 单品缓存：key为
product:{id}，缓存30分钟，后台修改商品时主动删除 - 缓存穿透保护：设置空值缓存（value为
__EMPTY__，过期时间60秒）
核心实现：
import json
import aioredis
from fastapi import Depends
redis = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_LIST_TTL = 300
CACHE_ITEM_TTL = 1800
EMPTY_MARK = "__EMPTY__"
async def get_products_with_cache(page: int, size: int):
cache_key = f"products:list:{page}:{size}"
# 1. 尝试读缓存
cached = await redis.get(cache_key)
if cached:
if cached == EMPTY_MARK:
return []
return json.loads(cached)
# 2. 查数据库（注意：这里需要异步驱动，我们用了asyncpg）
async with async_session() as session:
stmt = (
select(Product)
.options(selectinload(Product.skus), selectinload(Product.category))
.offset((page - 1) * size)
.limit(size)
)
result = await session.execute(stmt)
products = result.scalars().all()
data = [p.to_dict() for p in products]
# 3. 写入缓存
if data:
await redis.set(cache_key, json.dumps(data, ensure_ascii=False), ex=CACHE_LIST_TTL)
else:
await redis.set(cache_key, EMPTY_MARK, ex=60)
return data
注意这里把数据库驱动换成了asyncpg，因为之前用psycopg2同步驱动会阻塞事件循环，高并发时反而更慢。换成异步驱动后，配合async_session（AsyncSession），整个请求链路都是异步的。
5. 踩坑记录：序列化与缓存一致性
这里踩了两个坑，值得单独拿出来说：
坑1：to_dict()返回datetime对象，json.dumps直接报错。 解决方式是自定义JSONEncoder：
from datetime import datetime, date
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (datetime, date)):
return obj.isoformat()
return super().default(obj)
# 使用
json.dumps(data, cls=CustomEncoder, ensure_ascii=False)
坑2：后台修改商品后，列表缓存还是旧的。 我们写了个装饰器，在商品更新接口中主动删除相关列表缓存：
def invalidate_product_cache(product_id: int):
# 删除单品缓存
await redis.delete(f"product:{product_id}")
# 删除所有列表缓存（用scan匹配前缀）
async for key in redis.scan_iter(match="products:list:*"):
await redis.delete(key)
但这有个问题：如果列表页很多，scan会阻塞。后来改成了加版本号：products:list:v2:{page}:{size}，修改商品时版本号+1，旧key自然过期。简单粗暴，实测够用。
6. 最终压测数据与总结
最终版本（异步SQLAlchemy + Redis缓存 + 版本号失效策略）压测结果：
| 指标 | 原始版 | SQL优化版 | 缓存版 |
|---|---|---|---|
| TPS | 180 | 640 | 2100 |
| P50 | 480ms | 150ms | 38ms |
| P95 | 1200ms | 420ms | 75ms |
| P99 | 2800ms | 620ms | 120ms |
| 数据库QPS | ~10800 | ~1920 | ~420 |
数据库负载降了20多倍，接口延迟降了23倍。大促当天跑完，监控面板上这条接口的曲线稳如老狗。
总结下来，API性能调优的顺序应该是：
- 先profiling，别瞎猜。cProfile + py-spy都能快速定位热点
- SQL优化优先，N+1查询是最大的坑，ORM的
selectinload/joinedload要熟练 - 缓存是最后手段，但注意缓存穿透、雪崩、一致性这三个经典问题
- 异步化是隐藏福利，如果代码是同步阻塞的，换async驱动往往有意外惊喜
最后想说一点：不要过度设计。我们这个接口一开始用joinedload也行，但数据量大了以后selectinload更稳（不会产生笛卡尔积）。缓存TTL设5分钟，是因为商品数据确实变化不频繁，如果实时性要求高的接口，缓存策略还得重新考虑。
希望这篇能给正在搞API性能优化的朋友一些参考。有问题欢迎评论区交流。
