1. 背景：一个"能用但很慢"的接口
上个月接手了一个用户画像聚合服务，基于FastAPI + SQLAlchemy 2.0 + PostgreSQL 14构建。核心接口/api/v1/user/profile需要聚合用户基础信息、最近30天订单统计、以及动态标签。线上环境（4C8G容器）压测时发现：
- 平均响应时间：258ms
- P99：1.2s（严重超标）
- QPS：620
业务方反馈"能用但很慢"，尤其是大促期间经常超时。我先用locust做了一轮基础压测，确认瓶颈不在网关层，于是开始系统排查。
2. 环境与工具链
先交代一下具体版本，方便大家复现：
Python: 3.11.4
FastAPI: 0.104.1
Uvicorn: 0.24.0 (worker=4)
SQLAlchemy: 2.0.23
PostgreSQL: 14.9 (max_connections=200)
Redis: 7.0.12
压测工具: locust 2.18.4 (200并发, 持续5分钟)
调优工具我主要用了两个：
- py-spy：无侵入式采样，直接对运行中的进程做火焰图分析，不需要改代码
- cProfile：配合
cProfile.Profile()做局部函数的精细分析
3. 第一步：用py-spy定位热点函数
启动服务后，用locust跑到100并发，然后立刻对worker进程采样：
# 找到worker的PID
ps aux | grep uvicorn | grep -v grep
# 采样30秒，生成火焰图
py-spy record -p 12345 -o profile.svg --duration 30
打开火焰图后，问题一目了然：
UserProfileRepository.get_user_profile占用了47%的CPU时间- 其中SQLAlchemy的
Session.execute占了30% - 而且有大量
lazy_load调用——典型的N+1查询
再配合cProfile做精细统计：
import cProfile
import pstats
from app.api.v1.endpoints import user_profile
profiler = cProfile.Profile()
profiler.enable()
# 手动调用一次接口逻辑
result = user_profile.get_profile_sync(user_id=12345)
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative').print_stats(20)
输出显示：
ncalls tottime percall cumtime percall filename:lineno(function)
1 0.042 0.042 0.187 0.187 user_profile.py:56(get_profile)
12 0.051 0.004 0.135 0.011 orm.py:1206(load)
5 0.028 0.006 0.089 0.018 base.py:190(execute)
12次lazy_load，每次平均11ms——这就是罪魁祸首。
4. 方案设计：三管齐下
定位到问题后，我设计了三个优化点：
4.1 干掉N+1：改写为显式JOIN查询
原来的代码用ORM的relationship懒加载，每个用户的订单都要单独查一次。改为一次性JOIN查询：
# 优化前：N+1查询，12次DB往返
async def get_user_profile_old(user_id: int):
user = await db.get(User, user_id)
orders = await user.orders # lazy load
tags = await user.tags # lazy load
return assemble(user, orders, tags)
# 优化后：单次JOIN，3次查询（user/orders/tags各一次）
from sqlalchemy import select, func
async def get_user_profile_new(user_id: int):
# 用户基础信息
user_stmt = select(User).where(User.id == user_id)
user = (await db.execute(user_stmt)).scalar_one()
# 最近30天订单统计 - 用聚合函数，避免加载全部订单行
order_stmt = (
select(
Order.user_id,
func.count(Order.id).label('order_count'),
func.sum(Order.amount).label('total_amount'),
)
.where(Order.user_id == user_id, Order.created_at >= now - timedelta(days=30))
.group_by(Order.user_id)
)
order_stats = (await db.execute(order_stmt)).mappings().first()
# 标签直接查关联表
tag_stmt = select(Tag.name).join(UserTag).where(UserTag.user_id == user_id)
tags = list((await db.execute(tag_stmt)).scalars())
return assemble(user, order_stats, tags)
4.2 引入Redis缓存：缓存订单统计+标签
订单统计和标签是变化频率低的数据，非常适合缓存。我加了30秒的TTL，用user_id做key：
import json
import redis.asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_TTL = 30 # 秒
async def get_cached_order_stats(user_id: int):
cache_key = f"user:order_stats:{user_id}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 缓存未命中，查数据库
order_stats = await query_order_stats_from_db(user_id)
# 写入缓存，过期时间30秒
await redis_client.set(cache_key, json.dumps(order_stats), ex=CACHE_TTL)
return order_stats
4.3 优化序列化：避免Pydantic的重复校验
发现一个隐藏坑：响应模型里对Decimal字段做了两次Decimal转换，Pydantic的validator也被调用了两次。我直接用了model_validate替代手动构造，并关闭了不需要的字段校验：
from pydantic import BaseModel, ConfigDict
class UserProfileOut(BaseModel):
model_config = ConfigDict(validate_assignment=False, extra='ignore')
user_id: int
username: str
order_count: int
total_amount: float
tags: list[str]
5. 踩坑与优化细节
坑1：selectinload并不总是最优解
一开始我想偷懒，直接用SQLAlchemy的selectinload替代lazy load：
stmt = select(User).options(selectinload(User.orders)).where(User.id == user_id)
结果压测发现，虽然减少了DB往返，但selectinload会生成额外的IN子查询，在orders表数据量大时反而更慢。最终改用显式聚合查询，从12次查询降到3次。
坑2：Redis连接池耗尽
刚开始直接用aioredis创建连接，没有用连接池。压测到500并发时，Redis报Connection pool exhausted。解决方案：
# 用连接池，min_connections=10, max_connections=50
redis_client = aioredis.from_url(
"redis://localhost:6379/0",
decode_responses=True,
max_connections=50,
socket_connect_timeout=2,
)
坑3：Uvicorn worker数与CPU核数不匹配
默认4个worker在8核机器上反而性能下降，因为上下文切换开销大。调整为--workers 6后，QPS提升了15%左右。
6. 压测数据对比
用locust做了三轮对比测试，每轮5分钟，200并发：
| 指标 | 优化前 | 优化后（JOIN+Redis） | 提升 |
|---|---|---|---|
| 平均响应时间 | 258ms | 54ms | 4.78x |
| P99延迟 | 1.2s | 180ms | 6.67x |
| QPS | 620 | 2934 | 4.73x |
| 数据库查询次数/请求 | 12 | 3 | 75%减少 |
| 数据库CPU使用率 | 85% | 32% | 53%下降 |
Redis命中率在压测期间保持在94%以上（因为同一批测试用户反复请求）。
7. 总结与建议
这次调优的核心经验：
- 不要盲目优化，先定位。py-spy的火焰图30秒就能给出答案，别用猜的
- ORM的便捷是有代价的。N+1查询是最常见的性能杀手，但
selectinload也不是万能药，复杂聚合场景直接写SQL更可控 - 缓存要分层。订单统计和标签是读多写少的数据，缓存收益最大；但用户基础信息变化频繁，不适合缓存
- 压测要多次。单次压测结果受JIT预热、数据库缓存影响很大，至少跑3次取中位数
最后提醒一下：如果你的接口响应时间超过200ms，先看一眼是不是N+1查询。这个案例中，仅修复N+1就贡献了60%的性能提升，缓存策略贡献了剩下40%。
代码仓库：github.com/yourname/api-perf-tuning-demo（包含完整的优化前后代码和压测脚本）
