1. 问题背景：一个报表接口的“慢性死亡”
事情是这样的：我们有个 GET /api/v1/sales/summary 接口，返回最近 30 天各区域的销售汇总。上线初期一切正常，但数据量涨到 200 万行后，监控开始频繁报警——数据库 CPU 持续 80%+，接口超时率 5%。
第一反应是加索引？但查看慢查询日志发现，问题不在单条 SQL 慢，而是查询次数太多：一次请求竟然触发了 47 条 SQL。典型的 N+1 问题。更离谱的是，同一个区域的热门数据，每次请求都重新聚合计算，完全没利用缓存。
2. 环境与版本：不要忽视细节
- Python 3.10.12
- FastAPI 0.104.1（Uvicorn 0.24.0，workers=4）
- SQLAlchemy 2.0.23 + asyncpg 0.29.0
- PostgreSQL 14.5（配置 shared_buffers=4GB, effective_cache_size=12GB）
- Redis 7.0（单机，maxmemory 2GB）
- 压测工具：Locust 2.18.4（4 台 worker 机器，每台 200 并发）
- 部署环境：8C16G 云主机，网络延迟内网 = now - timedelta(days=30))
)
total = await db.scalar(stmt)
return {"total": total}
**踩坑提醒**：`selectinload` 适合一对多，`joinedload` 适合多对一。用错会导致笛卡尔积——我一开始用 `joinedload` 加载 2000 条订单，结果返回了 2000*区域数 行，直接内存溢出。
## 5. 缓存策略：Redis 与本地缓存两级方案
优化完 SQL 后，单次接口耗时降到 85ms（好多了），但压测发现 QPS 只到 700，**瓶颈变成了 PostgreSQL 的 CPU**——即使只有一条聚合查询，200 万行 group by 依然贵。
于是上缓存。方案设计：
- **一级缓存**：进程内 LRU（`functools.lru_cache`），TTL 60s，应对同一 worker 的重复请求
- **二级缓存**：Redis，TTL 300s，应对多 worker 共享
- **缓存键**：`sales:summary:{region}:{date}`，其中 date 是业务日期（避免缓存穿透）
核心实现：
```python
from functools import lru_cache
import aioredis
import json
# 一级缓存：注意 FastAPI 的协程需要加 async 版本
@lru_cache(maxsize=256)
def _get_cached_sync(region: str, date_str: str) -> str | None:
"""同步函数做本地缓存，因为 lru_cache 不支持 async"""
return None # 这里我们直接用 async 版本，见下方
# 实际用法：手动检查本地 + Redis
async def get_sales_summary_cached(region: str, date_str: str):
# 1. 本地缓存（用字典模拟，实际可用 cachetools）
cache_key = f"sales:summary:{region}:{date_str}"
if cache_key in local_cache:
return local_cache[cache_key]
# 2. Redis 缓存
redis_key = f"api_cache:{cache_key}"
cached = await redis.get(redis_key)
if cached:
data = json.loads(cached)
local_cache[cache_key] = data # 回填本地
return data
# 3. 数据库查询（此时已优化为单条聚合 SQL）
data = await query_db(region, date_str)
# 4. 写缓存：注意加随机过期时间，避免雪崩
ttl = 300 + random.randint(0, 60)
await redis.set(redis_key, json.dumps(data), ex=ttl)
local_cache[cache_key] = data
return data
踩坑：lru_cache 不能直接装饰 async 函数，会报错。我一开始直接 @lru_cache 在一个 async def 上，运行时报 TypeError: unhashable type: 'list'（参数是 pydantic 模型）。解决方案是：只缓存纯字符串参数，或者用 cachetools.TTLCache 的异步版本。
6. 压测数据与最终效果
用 Locust 写脚本，模拟真实用户行为：30% 请求华南，30% 华东，其余随机。压测 5 分钟，逐步增加并发。
# locustfile.py
from locust import HttpUser, task, between
class SalesUser(HttpUser):
wait_time = between(0.5, 2)
@task(3)
def summary_east(self):
self.client.get("/api/v1/sales/summary?region=华东")
@task(2)
def summary_south(self):
self.client.get("/api/v1/sales/summary?region=华南")
压测结果对比（200 并发，持续 5 分钟）：
| 指标 | 优化前 | 优化后（SQL） | 优化后（SQL+缓存） |
|---|---|---|---|
| QPS | 420 | 780 | 1350 |
| P50 延迟 | 280ms | 120ms | 40ms |
| P99 延迟 | 1.2s | 350ms | 280ms |
| PostgreSQL CPU | 82% | 45% | 12% |
| 单请求 SQL 次数 | 47 | 1 | 0（缓存命中） |
| 缓存命中率 | 0% | 0% | 94% |
注意：P99 并没有降到 40ms，因为缓存失效时有 6% 的请求会穿透到数据库，那部分耗时约 300ms。这符合预期——我们刻意让缓存 TTL 只有 5 分钟，保证数据新鲜度。
7. 总结：调优的顺序和原则
- 先 profiling，再优化。我见过太多人上来就加 Redis，结果的问题是 N+1 查询，缓存了也没用。
- 数据库优化收益最大。把 47 条 SQL 变成 1 条，比任何缓存都有效。
- 缓存是最后手段，不是银弹。我们加了缓存后，QPS 上来，但如果有写入场景，还要考虑一致性问题——目前用 TTL 5 分钟兜底，业务上可接受。
- 注意 Python 的 GIL：Uvicorn 多 worker 下，进程内缓存命中率只有 1/4（每个 worker 各存一份）。如果追求更高命中率，可以改用 Redis 单级缓存，但会增加 1-2ms 网络开销。
最后留个坑：这个服务还有另一个接口 GET /api/v1/sales/detail，它更麻烦——需要分页，且用户会频繁翻页。缓存键设计不好会导致内存爆炸。下次有空写写这个。如果这篇文章对你有帮助，欢迎收藏，有问题评论区交流。