1. 问题背景：迁移框架后，性能提升远低于预期
上个月，我把一个核心订单查询接口从Flask 2.2迁移到了FastAPI 0.104，预期性能能翻倍。结果压测数据让人尴尬：QPS从380涨到430，P95延迟只从890ms降到820ms。这12%的性能提升，基本是uvicorn异步worker带来的，跟FastAPI本身关系不大。
真正的问题在哪？我用py-spy dump看了一眼运行中的进程，发现json.dumps和SQLAlchemy ORM的序列化占了60%以上的CPU时间。也就是说，瓶颈根本不在框架，而在业务代码和数据库访问模式。
这篇文章不讲泛泛而谈的“优化三板斧”，只记录我实际做的三件事：Profiling定位热点 → SQL重写与索引优化 → 多级缓存策略。每一步都有压测数据支撑。
2. 环境与版本：先交代清楚，避免“版本玄学”
- Python 3.10.12（注意：3.11的字典和f-string有优化，但公司线上还是3.10）
- FastAPI 0.104.1 + Uvicorn 0.24.0（workers=4，loop="uvloop"，http="httptools"）
- Flask 2.2.3 + Gunicorn 20.1（gthread，workers=4，threads=8）
- SQLAlchemy 2.0.23（ORM模式）+ PostgreSQL 14.5
- Redis 6.2.6（单机，maxmemory-policy allkeys-lru）
- 压测工具：Locust 2.18，300并发，持续10分钟
测试环境是4核8G的云主机，数据库和Redis在同一内网，网络延迟约0.4ms。
3. Profiling工具对比：Py-Spy vs cProfile vs FastAPI内置
先说结论：cProfile对FastAPI这种异步框架会漏掉协程内的调用，而py-spy是采样式的，对生产环境无侵入，强烈推荐。
我用py-spy record --duration 30 --pid --format flamegraph -o flame.html抓了30秒火焰图。结果非常直观：
sqlalchemy.engine.base.Engine._execute_context占了38%的CPUpydantic.validators序列化占22%（这个是我用了response_model=OrderModel导致的）- 业务逻辑里的
for循环拼接SQL条件占了15%
优化动作1：去掉FastAPI的response_model，改用手动dict返回。 别误会，response_model是好东西，但它会对每个字段做校验和序列化，在QPS超过500时这开销不可忽略。我改用return order.to_dict()，只保留pydantic的输入校验。
# 优化前：response_model会导致二次序列化
@app.get("/api/orders/{order_id}", response_model=OrderResponse)
async def get_order(order_id: int):
order = await fetch_order_from_db(order_id)
return order
# 优化后：直接返回dict，减少一次pydantic序列化
@app.get("/api/orders/{order_id}")
async def get_order(order_id: int):
order = await fetch_order_from_db(order_id)
return order.to_dict() # to_dict里自己控制字段类型
这一项单独压测，P95从820ms降到710ms，约13%的提升。
4. 数据库查询优化：EXPLAIN ANALYZE揪出“罪魁祸首”
火焰图里_execute_context占比最高，说明SQL本身有问题。我打开慢查询日志（log_min_duration_statement = 200），发现一个查询执行了380ms：
SELECT * FROM orders
WHERE user_id = $1
AND status IN ('pending', 'paid')
ORDER BY created_at DESC
LIMIT 20;
表orders有500万行，user_id上有普通B-tree索引，但status没有索引。用EXPLAIN ANALYZE看执行计划：
Seq Scan on orders (cost=0.00..104218.40 rows=1234 width=248) (actual time=12.384..342.811 rows=987 loops=1)
Filter: ((status = 'pending'::text) OR (status = 'paid'::text))
Rows Removed by Filter: 4998761
问题很明确：user_id索引过滤后，回表再过滤status，导致大量随机IO。
优化方案：创建复合索引(user_id, status, created_at DESC)。注意created_at要放在最后，并且用DESC匹配ORDER BY。
CREATE INDEX CONCURRENTLY idx_orders_user_status_created
ON orders (user_id, status, created_at DESC);
同时我把查询从ORM改成了原生SQL（SQLAlchemy Core），因为ORM的select()在复杂查询时会生成多余的子查询：
# 优化后：使用SQLAlchemy Core，避免ORM一层开销
from sqlalchemy import text
async def fetch_orders(user_id: int, statuses: tuple):
sql = text("""
SELECT id, order_no, amount, status, created_at
FROM orders
WHERE user_id = :uid AND status IN :statuses
ORDER BY created_at DESC
LIMIT 20
""")
rows = await db.execute(sql, {"uid": user_id, "statuses": statuses})
return [dict(row._mapping) for row in rows]
效果：这条SQL从380ms降到42ms，查询时间减少88%。 整体接口P95降到480ms。
5. 缓存策略：Redis三级缓存，命中率与一致性
数据库优化后，接口P95是480ms，但QPS只有1050。因为数据库连接池满了，大量请求在等连接。这时候上缓存。
我设计了三层缓存，但这三层不是固定不变的，而是按数据热点动态切换：
- L1：进程内缓存（Python dict + 过期时间），TinyLRU算法，容量1000条，TTL 5秒。适合超高QPS的同一用户重复查询。
- L2：Redis缓存，key设计为
order:user:{user_id}:recent，value为JSON数组，TTL 60秒。 - L3：数据库，作为最终一致性兜底。
核心实现用cachetools库的TTLCache做L1，Redis做L2：
from cachetools import TTLCache
import json, redis.asyncio as aioredis
l1_cache = TTLCache(maxsize=1000, ttl=5) # 进程内缓存
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
async def get_orders_cached(user_id: int):
# L1: 进程内缓存
cache_key = f"user_orders:{user_id}"
if cache_key in l1_cache:
return l1_cache[cache_key]
# L2: Redis缓存
redis_key = f"order:user:{user_id}:recent"
cached = await redis_client.get(redis_key)
if cached:
data = json.loads(cached)
l1_cache[cache_key] = data # 回填L1
return data
# L3: 数据库查询
orders = await fetch_orders(user_id, ('pending', 'paid'))
data = [o for o in orders] # 假设每个o是dict
await redis_client.setex(redis_key, 60, json.dumps(data))
l1_cache[cache_key] = data
return data
踩坑点： 第一次我只加了Redis，没加进程缓存，压测发现QPS到1500就上不去了，因为每个请求都要走一次Redis网络IO（即使Redis很快，0.4ms延迟在300并发下也会放大）。加了L1后，进程内命中率约40%（压测场景下用户重复查询多），QPS直接拉到1870。
一致性处理： 订单状态变更时，用Redis的delete删除L2缓存，同时用l1_cache.pop(cache_key, None)清除L1。注意L1的TTL只有5秒，极端情况下最多5秒的脏读，可接受。
6. 最终压测数据与总结
在300并发、持续10分钟的压测下，优化前后对比：
| 指标 | Flask基线 | FastAPI迁移后 | Profiling+SQL优化后 | 加缓存后 |
|---|---|---|---|---|
| QPS | 380 | 430 | 1050 | 1870 |
| P50延迟 | 620ms | 580ms | 210ms | 88ms |
| P95延迟 | 890ms | 820ms | 480ms | 195ms |
| P99延迟 | 1.2s | 1.05s | 680ms | 310ms |
| CPU占用 | 85% | 80% | 65% | 42% |
总体提升：QPS提升391%，P95延迟降低76%。
几个值得说的结论：
- 别迷信框架迁移。 FastAPI确实比Flask快，但只占5%-10%的差距，真正的瓶颈在IO和序列化。
- Profiling工具用py-spy，别用cProfile。 cProfile在异步代码里会漏数据，而且对生产环境影响大。
- ORM不是万能的。 简单CRUD用ORM没问题，但复杂查询建议用SQLAlchemy Core或原生SQL，省掉一层对象映射。
- 缓存一定要分两级。 只加Redis的话，网络IO会成为新的瓶颈。进程内LRU缓存是吞吐量的关键。
最后，别过度优化。如果你的接口QPS不到500，把数据库索引加好、加一层Redis就足够了。我这套方案里L1进程缓存带来的复杂度（缓存失效、多worker内存不一致）只有在QPS超过1500时才值得承担。
