1. 问题背景：生产环境API响应时间飙到850ms
接手一个内部数据看板项目，FastAPI + SQLAlchemy + PostgreSQL。其中一个关键接口/api/dashboard/summary，返回用户最近30天订单统计。线上环境用户数量约5000并发，接口P99响应时间达到了850ms，导致前端加载超时。
初步排查：接口逻辑很简单，先查询用户基础信息，再查订单表做聚合。数据库每秒查询量约200，CPU使用率不高。直觉告诉我问题在应用层——ORM使用不当导致大量重复查询。
环境与版本：
- Python 3.12.2
- FastAPI 0.110.0
- SQLAlchemy 2.0.27
- asyncpg 0.29.0
- PostgreSQL 16.2
- Redis 7.2.4
- 服务器：4C8G ECS，CentOS 7
2. 工具选型与Profiling定位
先上工具。Python原生cProfile虽然准确，但FastAPI异步接口需要特殊处理。我选择用py-spy + cProfile组合，同时用fastapi-profiling中间件做采样。
Profiling步骤：
# 安装py-spy
pip install py-spy==0.3.14
# 启动FastAPI服务
uvicorn main:app --workers 2
# 另一个终端运行压测（压测工具用wrk）
wrk -t4 -c50 -d30s http://localhost:8000/api/dashboard/summary
# 采样火焰图
sudo py-spy record -o profile.svg --pid $(pgrep -f uvicorn) --duration 30
火焰图结果令我震惊：session.execute 调用占用了超过65%的CPU时间，而其中_fetch_all_rows方法反复执行了3000多次（单次请求）。代码逻辑里明明只有一条SQL查询，为何执行3000次？
真相：SQLAlchemy的relationship懒加载（lazy='select'）导致每次访问关联属性都触发一次查询。典型N+1问题。配合SQLALCHEMY_ECHO=1打印SQL日志，发现一次API请求触发了150条SQL语句。
3. 数据库查询优化：从N+1到JOIN + 索引
核心代码优化前（有问题的版本）：
# app/services/dashboard.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models import User, Order
async def get_dashboard_summary(db: AsyncSession, user_id: int):
# 1次查询用户
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
return None
# N次查询每个用户的订单（当user.orders是lazy加载时）
# 这里user.orders会触发额外的查询
orders = user.orders # 触发懒加载
total_amount = sum(order.amount for order in orders)
order_count = len(orders)
return {"total_amount": total_amount, "order_count": order_count}
问题：user.orders触发了懒加载。如果用户有150个订单，就会执行150次SQL查询。
优化方案（使用eager loading + 索引）：
# app/services/dashboard.py
from sqlalchemy.orm import selectinload
from sqlalchemy import func, select
async def get_dashboard_summary_optimized(db: AsyncSession, user_id: int):
# 使用selectinload一次性加载关联订单
stmt = (
select(User)
.options(selectinload(User.orders))
.where(User.id == user_id)
)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if not user:
return None
# 内存中计算聚合
total_amount = sum(order.amount for order in user.orders)
order_count = len(user.orders)
return {"total_amount": total_amount, "order_count": order_count}
索引优化：在orders表的user_id上增加复合索引，覆盖查询字段：
CREATE INDEX idx_orders_user_id_amount ON orders (user_id, amount);
踩坑：selectinload在SQLAlchemy 2.0中默认使用SELECT IN而不是JOIN，这在高并发场景下对数据库压力更小。但要注意，如果关联数据量超过1000条，建议改用joinedload。我测试后发现用户订单通常不超过500条，selectinload表现更好。
4. 缓存策略：多级缓存 + 失效更新
即使优化了查询，每次请求都查数据库仍然不够快。对于这种写少读多的聚合数据，缓存是必选项。
设计思路：
- 一级缓存：应用内内存缓存（cachetools，TTL=60秒）
- 二级缓存：Redis缓存（TTL=300秒）
- 失效策略：当用户新增订单时，主动失效缓存
核心实现：
# app/cache.py
from cachetools import TTLCache
import aioredis
from typing import Optional
# 一级内存缓存，最大1000条，TTL 60秒
memory_cache = TTLCache(maxsize=1000, ttl=60)
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
async def get_cached_summary(user_id: int) -> Optional[dict]:
# 先查内存缓存
cache_key = f"dashboard_summary:{user_id}"
if cache_key in memory_cache:
return memory_cache[cache_key]
# 再查Redis缓存
cached = await redis_client.get(cache_key)
if cached:
import json
data = json.loads(cached)
memory_cache[cache_key] = data # 回填内存
return data
return None
async def set_cached_summary(user_id: int, data: dict):
import json
cache_key = f"dashboard_summary:{user_id}"
memory_cache[cache_key] = data # 更新内存
await redis_client.setex(cache_key, 300, json.dumps(data)) # Redis TTL 5分钟
async def invalidate_cache(user_id: int):
cache_key = f"dashboard_summary:{user_id}"
memory_cache.pop(cache_key, None)
await redis_client.delete(cache_key)
在FastAPI路由中使用：
# app/routers/dashboard.py
from fastapi import APIRouter, Depends
from services.dashboard import get_dashboard_summary_optimized
from cache import get_cached_summary, set_cached_summary
router = APIRouter()
@router.get("/api/dashboard/summary")
async def dashboard_summary(
user_id: int,
db: AsyncSession = Depends(get_db)
):
# 尝试缓存
cached = await get_cached_summary(user_id)
if cached:
return cached
# 缓存未命中，查询数据库
data = await get_dashboard_summary_optimized(db, user_id)
await set_cached_summary(user_id, data)
return data
踩坑：Redis连接池配置。默认aioredis.from_url会创建连接池，但如果不设置max_connections，默认只有10个连接。高并发下会导致连接等待。建议设置为max_connections=50。
5. 压测数据：优化前后对比
使用Locust 2.20.1做压测，设置1000并发用户，持续3分钟。
压测脚本关键参数：
# locustfile.py
from locust import HttpUser, task, between
class DashboardUser(HttpUser):
wait_time = between(0.5, 2)
@task
def get_summary(self):
self.client.get("/api/dashboard/summary", params={"user_id": 12345})
结果对比：
| 指标 | 优化前 | 优化后(仅SQL) | 优化后(SQL+缓存) |
|---|---|---|---|
| 平均响应时间 | 850ms | 120ms | 45ms |
| P99响应时间 | 1.2s | 280ms | 98ms |
| 吞吐量 (req/s) | 45 | 320 | 850 |
| 数据库CPU | 65% | 22% | 5% |
| SQL执行次数/请求 | 150 | 1 | 0 (缓存命中) |
缓存命中率：在持续压测下，缓存命中率稳定在92%左右。因为压测用的user_id是固定的，实际生产环境需要根据业务调整TTL。
6. 总结与经验
- Profiling优先，不要拍脑袋优化。py-spy火焰图让我15分钟就定位了N+1问题，省去了大量猜测时间。
- ORM的懒加载是性能杀手。SQLAlchemy 2.0默认lazy='select'，生产环境必须显式使用
selectinload或joinedload。 - 缓存不是银弹，但高并发下是必选项。两级缓存设计（内存+Redis）有效降低数据库压力。注意缓存失效策略，避免数据不一致。
- 索引不要贪多。我只给order表加了user_id和amount的复合索引，覆盖了查询和聚合字段。过多索引会影响写入性能。
- 参数调优：PostgreSQL的
shared_buffers设为2GB（总内存的25%），work_mem设为64MB。Redis的maxmemory设为1GB，淘汰策略allkeys-lru。
最终效果：API从850ms降至45ms，数据库CPU从65%降到5%。服务器从4C8G还能再支撑3倍流量。
以上是我这次调优的全过程，代码可以直接复制使用。如果你有更好的方案，欢迎评论区交流。
