1. 背景：从“能用”到“撑不住”的300ms接口
半年前接手一个商品聚合查询API，技术栈为FastAPI 0.78 + SQLAlchemy 1.4 + PostgreSQL 13 + Redis 6.2。接口逻辑很简单：根据品类ID返回商品列表及关联的SKU、标签、库存信息。压测时发现，单核4GB的容器下，并发200时延迟从300ms暴涨至850ms，CPU冲到95%。
初步怀疑是数据库问题，但查看慢查询日志，单条SQL都在10ms以内。真正的罪魁祸首藏在应用层——一个看似无害的循环。
2. 环境与版本
| 组件 | 版本 | 配置 |
|---|---|---|
| Python | 3.9.12 | Gunicorn + Uvicorn workers=4 |
| FastAPI | 0.78.0 | 无特殊配置 |
| SQLAlchemy | 1.4.36 | sessionmaker(autocommit=False) |
| PostgreSQL | 13.4 | shared_buffers=1GB, effective_cache_size=3GB |
| Redis | 6.2.6 | maxmemory=2GB, eviction=allkeys-lru |
| locust | 2.12.1 | 压测客户端 |
3. 第一刀：用py-spy抓住“看不见的循环”
常规的cProfile对异步代码支持差，我用py-spy在压测时实时采样：
# 安装
pip install py-spy
# 采样30秒，生成火焰图
py-spy record -o flame.svg -p $(pgrep -f uvicorn) --duration 30
火焰图清晰显示：36% 的时间花在sqlalchemy.orm.loading._instance_processor，28% 花在json.dumps，15% 花在decimal.Decimal.__repr__。
定位到代码：
# 原始代码：每次访问关系属性都会触发新查询
products = session.query(Product).filter(Product.category_id == cat_id).all()
for product in products:
product.skus # 这里！触发N+1
product.tags # 这里！又触发一次
每个product访问.skus时，SQLAlchemy惰性加载会发一条SELECT * FROM skus WHERE product_id = ?。接口返回50个商品，就多出50+条SKU查询和50+条标签查询，总计98条SQL。
4. 数据库查询优化：Eager Loading + 批量查询
改造后的查询：
from sqlalchemy.orm import selectinload
def get_products_with_relations(cat_id: int):
query = (
session.query(Product)
.options(
selectinload(Product.skus), # 一次IN查询加载所有SKU
selectinload(Product.tags),
selectinload(Product.inventory),
)
.filter(Product.category_id == cat_id)
)
return query.all()
selectinload会将关系查询合并为SELECT * FROM skus WHERE product_id IN (:id1, :id2, ...)。压测对比：
| 指标 | 改造前 | 改造后 |
|---|---|---|
| 单请求SQL数量 | 98条 | 4条（1主+3子查询） |
| 数据库连接池占用 | 频繁等待 | 稳定在8-12连接 |
| P50延迟 | 312ms | 87ms |
| P99延迟 | 850ms | 210ms |
踩坑记录：selectinload会生成IN子句，但PostgreSQL对IN列表超过1000个元素时性能下降。我实测在500个元素以内性能线性增长，超过后改用subqueryload。
5. 缓存策略：二级缓存扛住热点数据
数据库优化后P99降到210ms，但高频接口（同一品类被大量用户查询）依然有压力。引入Redis二级缓存：
# 缓存策略：一级进程缓存（lru_cache）+ 二级Redis
from functools import lru_cache
import json
import aioredis
redis = aioredis.from_url("redis://localhost:6379/1", decode_responses=True)
class ProductCache:
# 一级缓存：进程内LRU，最多缓存200个品类
@lru_cache(maxsize=200)
def _local_cache(self, cat_id: int) -> str:
return None # 占位，实际逻辑在get方法中
async def get(self, cat_id: int) -> list[dict]:
# 1. 读本地缓存（本地字典，无序列化开销）
local_key = f"cat:{cat_id}"
if local_key in self._local_cache.cache:
return self._local_cache.cache[local_key]
# 2. 读Redis
redis_key = f"products:cat:{cat_id}"
cached = await redis.get(redis_key)
if cached:
data = json.loads(cached)
self._local_cache.cache[local_key] = data
return data
# 3. 查数据库
products = get_products_with_relations(cat_id)
data = serialize_products(products)
# 4. 写入Redis，TTL=60秒
await redis.setex(redis_key, 60, json.dumps(data))
self._local_cache.cache[local_key] = data
return data
缓存命中率监控：用redis-cli info stats观察，命中率稳定在92%-95%。注意点：一级缓存使用lru_cache但未设置过期时间——因为二级缓存已控制TTL，一级缓存仅作为热数据加速。
6. JSON序列化：被忽略的28%耗时
火焰图中json.dumps占28%，排查发现原因是Decimal类型。SQLAlchemy返回的price字段是Decimal('99.99')，原生json模块无法序列化，我们之前用了自定义encoder：
# 原始自定义编码器，性能极差
class OldEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj) # 转换一次
return super().default(obj)
每次调用float(obj)会触发Python的Decimal→float转换，且JSONEncoder每次只处理一个对象。换成批量预处理：
# 优化方案：批量预处理数值字段
from decimal import Decimal
import orjson # 比标准库快4-6倍
def serialize_products(products: list) -> list[dict]:
result = []
for p in products:
d = p.__dict__.copy()
d.pop('_sa_instance_state', None)
# 批量转换所有Decimal字段
for field in ['price', 'original_price', 'discount']:
if isinstance(d.get(field), Decimal):
d[field] = float(d[field])
result.append(d)
return result
# API中使用orjson
@app.get("/products/{cat_id}")
async def get_products(cat_id: int):
data = await cache.get(cat_id)
return orjson.dumps(data).decode()
性能对比（压测数据）：
| 序列化方式 | 平均耗时 | 吞吐量 |
|---|---|---|
| json.dumps + OldEncoder | 28ms | 1200 QPS |
| json.dumps + 预处理 | 9ms | 3400 QPS |
| orjson + 预处理 | 3ms | 5600 QPS |
最终采用orjson，序列化耗时从28ms降至3ms。
7. 压测与最终效果
用locust做压测，配置如下：
# locustfile.py
from locust import HttpUser, task, between
class APIUser(HttpUser):
wait_time = between(0.1, 0.5)
@task
def get_products(self):
# 模拟真实场景：70%请求热门品类，30%随机
cat_id = 1 if self.random.random() < 0.7 else self.random.randint(1, 100)
self.client.get(f"/products/{cat_id}")
最终数据对比（并发200，运行5分钟）：
| 指标 | 优化前 | 优化后 | 提升倍数 |
|---|---|---|---|
| P50延迟 | 312ms | 9ms | 34.7x |
| P99延迟 | 850ms | 18ms | 47.2x |
| QPS | 1200 | 9600 | 8.0x |
| CPU使用率 | 95% | 42% | -53% |
8. 总结与踩坑清单
- Profiling第一：别信直觉，用py-spy火焰图找到真瓶颈。我浪费了两天优化数据库索引，结果问题是N+1和序列化。
- SQLAlchemy N+1：
selectinload适合关联数<500的场景，超过用subqueryload。注意joinedload可能会产生笛卡尔积。 - 缓存层级：本地缓存+Redis二级，本地缓存不要设过期（让Redis控制），否则会造成缓存击穿。
- JSON序列化：别用标准库的
default回调，批量预处理数值类型，然后上orjson。 - 压测模拟真实流量：热点数据占比要接近生产，否则缓存策略测不准。
最后送一句：API优化不是堆硬件，而是用profiling找到那根最粗的瓶颈，然后一刀砍下去。
