一、问题背景：一个“简单”报表接口为什么这么慢
事情是这样的，上周三下午，运营同学反馈数据看板打开要转圈5秒以上。我打开监控面板，发现/api/v1/report/summary这个接口的P95响应时间已经飙到4380ms，而数据库CPU使用率长期徘徊在90%左右。
这个接口逻辑并不复杂：从订单表、用户表、商品表各取一些聚合数据，再拼装成一个JSON返回。当时想当然地认为“就是几个count和sum，能慢到哪去”，直到我用工具一看，发现事情并不简单。
环境版本（方便你复现）：
- Python 3.10.12
- FastAPI 0.104.1
- Uvicorn 0.24.0
- SQLAlchemy 2.0.23
- Redis 7.0.12
- PostgreSQL 15.3
二、第一步：profile定位，别瞎猜
很多人遇到性能问题第一反应就是“加缓存”或者“改索引”，我劝你先别急。我习惯先用cProfile快速扫一遍，再用py-spy抓线上真实调用栈。
本地profiling：
# 用cProfile跑一次请求，输出到文件
python -m cProfile -o report.prof main.py
# 用snakeviz可视化
snakeviz report.prof
打开snakeviz后，我立刻看到了罪魁祸首——sqlalchemy.orm.session.execute占了总耗时的72%，而且被调用了2400多次。
接着用py-spy抓线上：
# 找出uvicorn worker的PID
pgrep -f uvicorn
# 抓取3秒的调用栈快照
py-spy dump --pid --duration 3
看到栈顶全是lazy load相关操作，心里就有数了——典型的N+1查询。代码里用了SQLAlchemy的relationship，但没配置加载策略，导致每次访问关联对象都发一条SQL。
三、第二步：数据库查询优化，先解决N+1
原来的代码长这样（简化版）：
# 问题代码：每个订单都触发一次用户查询
async def get_summary():
orders = db.query(Order).filter(Order.created_at >= yesterday).all()
for order in orders:
order.user.name # 这里触发N+1！
order.items # 这里又触发N+1！
优化方案：用SQLAlchemy 2.0的selectinload一次性加载所有关联对象，同时对聚合查询改用func.count()一次性算完。
# 优化后：selectinload + 聚合查询
from sqlalchemy.orm import selectinload
from sqlalchemy import func, select
async def get_summary_optimized():
# 方式1：一条JOIN查完所有需要的关联
stmt = (
select(Order)
.options(selectinload(Order.user))
.options(selectinload(Order.items))
.where(Order.created_at >= yesterday)
)
orders = await db.execute(stmt)
# 方式2：聚合直接用SQL算，不加载到Python
agg_stmt = (
select(
func.count(Order.id),
func.sum(Order.amount),
func.count(func.distinct(Order.user_id))
)
.where(Order.created_at >= yesterday)
)
result = await db.execute(agg_stmt)
效果对比（本地wrk压测，并发100，持续30秒）：
| 指标 | 优化前 | 优化后 |
|---|---|---|
| 查询次数 | 2400+ | 3 |
| P95延迟 | 4380ms | 950ms |
| 数据库CPU | 90% | 35% |
四、第三步：缓存策略，该上就上
查询优化后P95降到950ms，但离目标还差得远。这个接口的数据是分钟级更新的报表汇总，完全没必要每次都查数据库。
我的缓存方案分两层：
1. Redis缓存（跨进程共享）：缓存key为report:summary:2024-01-15，TTL设120秒。
2. 进程内LRU缓存（本机最快）：用functools.lru_cache做第一层兜底，TTL控制在10秒内，避免多worker间数据不一致太严重。
代码实现：
import json
import redis
from functools import lru_cache
r = redis.Redis(host='localhost', port=6379, db=0)
# 进程内LRU：最多缓存256个key，10秒过期
@lru_cache(maxsize=256)
def get_report_from_local(date_str: str):
return None # 占位，实际返回None表示未命中
async def get_summary_with_cache(date_str: str):
# 第一层：本地LRU
local_key = f"report:{date_str}"
cached = get_report_from_local(local_key)
if cached:
return json.loads(cached)
# 第二层：Redis
redis_key = f"report:summary:{date_str}"
cached = r.get(redis_key)
if cached:
# 更新本地缓存
get_report_from_local.__wrapped__(local_key, cached)
return json.loads(cached)
# 第三层：查数据库（已优化过的查询）
data = await compute_summary_from_db(date_str)
# 写双级缓存
data_str = json.dumps(data)
r.setex(redis_key, 120, data_str) # Redis 120秒过期
get_report_from_local.__wrapped__(local_key, data_str) # 本地10秒
return data
踩坑提醒：
1. lru_cache在FastAPI异步环境下要小心——不能直接装饰async函数，我这里是同步函数包了一层。
2. Redis的setex的TTL不要太长，否则数据过期会很突兀。我这边业务能接受2分钟延迟，所以设120秒。
3. 别忘了缓存穿透——如果数据库里没数据，Redis会缓存None，我额外加了cache_null标志位处理。
五、第四步：压测验证与调参
用wrk压测最终版本，服务器配置：4核8G，PostgreSQL和Redis都在同一台机器（测试环境）。
wrk -t8 -c200 -d60s --latency http://localhost:8000/api/v1/report/summary
最终结果对比：
| 指标 | 原始版本 | 查询优化 | 查询+缓存 |
|---|---|---|---|
| QPS | 45 | 210 | 1800 |
| P95 | 4380ms | 950ms | 210ms |
| 数据库CPU | 90% | 35% | 15% |
| 内存占用 | - | - | +120MB（Redis） |
六、总结与思考
这次调优下来，最大的感受是先定位再优化。如果一开始就盲目加缓存，可能掩盖了N+1这个真正的问题，缓存一旦失效，性能会恶化得更厉害。
最后复盘一些关键决策：
- profile工具选择：cProfile适合本地，py-spy适合线上，两个配合用基本能定位90%的问题。
- 缓存层级设计：Redis解决跨进程共享，进程内LRU解决热点访问，两层配合能扛住突发流量。
- 数据一致性：我用了简单的TTL策略，如果业务要求更强的一致性，可以考虑Redis发布订阅做主动失效。
如果你也在调FastAPI/Flask的接口性能，建议先跑一遍profile，看看时间到底花在哪，别急着上缓存。数据库查询优化通常是性价比最高的第一步。
最后留两个问题供大家思考：
1. 如果你的API是Flask而不是FastAPI，上面的异步方案要怎么调整？
2. 当Redis本身成为瓶颈时，你会怎么优化？
