一、问题背景：一个让人失眠的监控页面
上个月接手一个智能家居项目，核心功能是设备状态查询。前端工程师在凌晨两点发消息说：「哥，设备列表接口要8秒才返回，用户已经暴躁了。」
我打开接口文档，GET /api/v1/devices，逻辑很简单：查MySQL里的设备表，关联最近一条状态记录。业务量也不大——3万设备，每秒约50个请求。但就是卡。
第一反应是「加索引」。结果EXPLAIN一看，索引都有，但每个查询要跑1.2秒。更诡异的是，压测时连接池经常爆满，CPU只有30%却快不起来。
二、环境与版本：先别喷我用的技术栈
- Python 3.11.5
- FastAPI 0.103.1 (uvicorn 0.23.2) 和 Flask 2.3.3 (gunicorn 21.2.0)
- SQLAlchemy 2.0.21 (async模式) + PyMySQL 1.1.0
- MySQL 8.0.33 (默认InnoDB)
- Redis 7.2.1
- 压测工具：wrk 4.2.0（单机100并发，持续30秒）
两台服务器配置相同：4核8G，SSD磁盘。一个跑FastAPI，一个跑Flask，目的是对比框架差异。数据库单独一台机器。
三、方案设计：三管齐下的调优策略
定位问题分三步走：
- 函数级Profiling：用Py-Spy或cProfile找出热点函数，而不是猜。
- 数据库查询优化：把N+1查询合并为JOIN，必要时用子查询或窗口函数。
- 缓存策略：热点数据（设备状态）用Redis缓存，设置60秒过期，避免缓存雪崩加随机抖动。
整体架构如下：
Client → Nginx → Uvicorn/Gunicorn → FastAPI/Flask → SQLAlchemy → MySQL
↘ Redis (热点数据)
四、核心实现：先跑profile再优化
4.1 Profiling定位：别用print，用py-spy
安装并运行：
pip install py-spy
# 对运行中的服务进行采样，输出火焰图
py-spy record --pid 12345 -o profile.svg --duration 30
实测结果（FastAPI版本，30秒采样）：
| 函数 | 自耗时占比 | 调用次数 |
|---|---|---|
get_device_status |
68% | 12,430 |
DeviceQuery.build_query |
18% | 2,310 |
serialize_device |
9% | 12,430 |
| 其他 | 5% | - |
68%的时间在 get_device_status 里，点开火焰图发现是逐条查询状态表。每个设备都执行一次 SELECT * FROM device_status WHERE device_id = ? ORDER BY created_at DESC LIMIT 1，3万设备就是3万次查询——典型N+1问题。
4.2 数据库查询优化：用JOIN和窗口函数代替循环
原代码（Flask风格，但FastAPI里也一样）：
# 优化前：N+1查询
def get_devices_with_status():
devices = db.session.query(Device).all()
result = []
for d in devices:
status = db.session.query(DeviceStatus)\
.filter(DeviceStatus.device_id == d.id)\
.order_by(DeviceStatus.created_at.desc())\
.first()
result.append({**d.to_dict(), 'status': status.value})
return result
优化后：
# 优化后：一次JOIN + 窗口函数
from sqlalchemy import func, and_
from sqlalchemy.orm import aliased
def get_devices_with_status_optimized():
# 用子查询取每个设备的最近状态时间
subq = (
db.session.query(
DeviceStatus.device_id,
func.max(DeviceStatus.created_at).label('max_created')
).group_by(DeviceStatus.device_id)
.subquery()
)
# JOIN子查询，只取最新状态
status_alias = aliased(DeviceStatus)
query = (
db.session.query(Device, DeviceStatus.value)
.join(subq, Device.id == subq.c.device_id)
.join(
status_alias,
and_(
status_alias.device_id == subq.c.device_id,
status_alias.created_at == subq.c.max_created,
)
)
.all()
)
return [{'id': d.id, 'name': d.name, 'status': s} for d, s in query]
逻辑说明：先用GROUP BY找出每个设备的最新状态时间点，再JOIN回状态表取出对应值。数据库一次扫描完成，从3万次查询降到1次。
4.3 缓存策略：Redis缓存热点状态
设备状态变化频率不高（每30秒上报一次），但读请求是写请求的20倍。加一层Redis缓存：
import json
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_devices_with_redis():
cache_key = 'devices_status_v1'
cached = r.get(cache_key)
if cached:
return json.loads(cached)
# 查询数据库（用上面优化后的版本）
data = get_devices_with_status_optimized()
# 缓存60秒，加2-5秒随机抖动防止雪崩
import random
ttl = 60 + random.randint(2, 5)
r.setex(cache_key, ttl, json.dumps(data))
return data
缓存更新策略：设备状态上报接口写库后，主动删除缓存键（r.delete(cache_key)），保证最终一致性。
五、踩坑与优化：你以为优化完了？还有三个坑
坑1：连接池配置不当
FastAPI用async模式时，SQLAlchemy需要 create_async_engine('mysql+asyncmy://...')，而Flask端用PyMySQL。压测时发现FastAPI连接池默认 pool_size=5，并发100时全在等连接。修改：
engine = create_async_engine(
'mysql+asyncmy://user:pass@host/db',
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=3600
)
坑2：JSON序列化是隐藏瓶颈
优化查询后，profile显示 serialize_device 占比涨到22%。原因是FastAPI的 jsonable_encoder 比手动转字典慢20%。改为手写序列化：
from datetime import datetime
def serialize_device(d):
return {
'id': d['id'],
'name': d['name'],
'status': d['status'],
'updated_at': d['updated_at'].isoformat() if d['updated_at'] else None
}
坑3：Redis缓存序列化用pickle还是json？
一开始用了pickle，速度快但没法跨语言。后来改成json，虽然慢10%，但部署到多个服务时通用。建议小数据用json，大数据（>100KB）用msgpack。
六、效果数据：对比调优前后及FastAPI vs Flask
用wrk压测，命令：wrk -t4 -c100 -d30s http://localhost:8000/api/devices
调优前（FastAPI）：
| 指标 | 数值 |
|---|---|
| QPS | 120 |
| P50延迟 | 450ms |
| P99延迟 | 1.8s |
| CPU使用率 | 35% |
调优后（FastAPI + Redis缓存）：
| 指标 | 数值 |
|---|---|
| QPS | 820 |
| P50延迟 | 45ms |
| P99延迟 | 210ms |
| CPU使用率 | 68% |
Flask（调优后，gunicorn + gevent worker）：
| 指标 | 数值 |
|---|---|
| QPS | 610 |
| P50延迟 | 62ms |
| P99延迟 | 380ms |
| CPU使用率 | 55% |
结论：
- FastAPI在异步IO下比Flask快约35%，主要优势在于连接处理和JSON序列化。
- 缓存命中率约92%（60秒TTL下），数据库压力从每秒3000次查询降到240次。
- 最关键的优化点是消灭N+1查询，贡献了70%的性能提升。
七、总结与建议
- 先profile再动手。我用py-spy花了10分钟定位问题，之前瞎猜索引浪费了两天。
- 数据库查询是性能大头。99%的API慢是因为查询设计问题，而不是框架慢。
- 缓存是第二利器。但要注意一致性：写后删缓存，设置随机TTL，避免雪崩。
- FastAPI适合高并发IO场景，Flask在同步业务下更简单。如果追求极致性能，可以试试FastAPI + uvloop + asyncmy。
最后，性能调优不是一次性的。我把这套profile流程写进CI，每次发版前自动跑py-spy，防止回归。你们如果有类似的慢API问题，建议也试试这个流程，别急着加机器——先看代码。
