一、问题背景：从Flask迁移到FastAPI后性能不升反降
去年接手一个订单查询服务，技术栈是Flask 2.2 + SQLAlchemy 1.4 + PostgreSQL 14。当时线上QPS 400，P99 900ms，勉强能跑。为了追求异步性能，我们决定迁移到FastAPI 0.104 + asyncpg。结果迁移完成后压测，QPS只有380，P99反而到了1.2s——比Flask还慢。
排查后发现两个问题：一是代码里全是同步SQLAlchemy调用，FastAPI的异步优势完全没用上；二是模板渲染用了Jinja2的render_template，每次请求渲染3KB的HTML竟然耗时300ms+。这让我意识到，技术栈迁移不能解决架构设计问题，必须从profiling开始。
二、环境与版本：压测工具与监控栈
- 服务端：Python 3.11.5，FastAPI 0.104.1，Flask 2.3.3，SQLAlchemy 2.0.21，Redis 7.0，PostgreSQL 14.5
- 压测工具：wrk 4.2.0（单线程100连接，持续60s）
- Profiling工具：PySpy 0.3.14（现场抓栈），py-spy record --duration 30 -o profile.json --pid
- 部署：Docker容器，4核8G，宿主机为同一台压测机（排除网络干扰）
这里有个坑：wrk必须和服务器在同一台机器或同一内网，否则网络延迟会淹没应用层优化效果。我见过太多人用本机wrk压远程服务器，结果P99全被网络吃掉。
三、方案设计：三层优化策略
- Profiling先行：用PySpy抓线上真实调用栈，定位CPU和IO热点，而不是靠猜。
- 数据库层：用
selectinload替代lazy load，减少SQL查询次数；对大表加覆盖索引。 - 缓存策略：对热点订单数据做Redis二级缓存，设置TTL 60s；对响应开启gzip压缩。
核心思路：先看瓶颈在哪，再决定优化方向。我见过太多人上来就加缓存，结果缓存命中率不到10%，还引入了缓存一致性问题。
四、核心实现：从PySpy到SQLAlchemy优化
4.1 PySpy定位瓶颈：现场抓栈
# 安装
pip install py-spy==0.3.14
# 抓取运行中的服务栈（PID替换为实际进程号）
py-spy record --duration 30 -o /tmp/profile.json --pid 12345
# 生成火焰图（需要flamegraph.pl）
py-spy dump --pid 12345 | grep -A 20 "Thread 0x"
抓栈结果让我震惊：render_template占了45%的CPU时间，SQLAlchemy的lazy load占了30%，剩下25%在JSON序列化。问题根本不在框架，而在SQL查询次数——每个订单详情要额外执行5次查询（用户、商品、物流、优惠券、评价），这就是典型的N+1问题。
4.2 SQLAlchemy优化：selectinload替代lazy load
# 优化前（Flask/SQLAlchemy 1.4风格）
order = db.session.query(Order).filter(Order.id == order_id).first()
# 每次访问 order.user 都会触发一条SQL
# 优化后（FastAPI/SQLAlchemy 2.0异步风格）
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import selectinload
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async with AsyncSession(engine) as session:
stmt = (
select(Order)
.options(
selectinload(Order.user),
selectinload(Order.items).selectinload(Item.sku),
selectinload(Order.logistics),
)
.where(Order.id == order_id)
)
result = await session.execute(stmt)
order = result.scalar_one()
这个改动把查询次数从6次降到1次（通过JOIN），SQL执行时间从80ms降到12ms。注意selectinload会生成IN子查询，对大结果集比joinedload更友好——后者会产生笛卡尔积。
4.3 Redis二级缓存：只缓存热点数据
import json
import aioredis
redis = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
async def get_order_with_cache(order_id: int):
cache_key = f"order:{order_id}:v2"
# 先查缓存
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# 查数据库
order = await fetch_order_from_db(order_id)
if order:
# 序列化后缓存，TTL 60秒
await redis.set(cache_key, json.dumps(order, default=str), ex=60)
return order
这里有个关键细节：不是所有订单都缓存。我们只缓存近30分钟内的订单（通过order.create_time > now - 1800s判断），冷数据直接查库。缓存命中率实测61%，但P99从1.2s降到150ms——因为热数据完全绕过数据库和模板渲染。
4.4 FastAPI中间件：gzip压缩 + 响应优化
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=500, compresslevel=5)
# 自定义响应模型，避免Jinja2渲染
from pydantic import BaseModel
class OrderResponse(BaseModel):
order_id: int
user_name: str
total_amount: float
items: list[dict]
@app.get("/api/order/{order_id}", response_model=OrderResponse)
async def get_order(order_id: int):
# 直接返回JSON，不走Jinja2
return await get_order_with_cache(order_id)
Jinja2模板渲染是CPU密集型操作，在4核机器上并发400时CPU直接打满。改成Pydantic序列化后，响应体从3KB降到800B（移除了HTML标签），gzip后网络传输只有120B。
五、踩坑与优化：asyncpg连接池与索引设计
三个坑必须说：
-
asyncpg连接池默认size=10，压测时连接池被打爆，大量请求等待连接。需要显式设置
pool_size=20, max_overflow=10，并开启pool_pre_ping=True防止死连接。 -
覆盖索引：订单表有1000万数据，
WHERE order_id = ?虽然走主键索引，但需要回表查询。我们加了复合索引(order_id, create_time) INCLUDE (status, total_amount)，查询只需扫索引页，IO从8KB降到2KB。 -
Redis序列化：用
json.dumps(default=str)会把Decimal转字符串，前端解析会出错。正确做法是default=lambda o: float(o) if isinstance(o, Decimal) else str(o)。
还有一个优化是异步化模板渲染——既然Jinja2跑不动，干脆不用模板，改为前端静态页面+JSON API。这个决策让响应体小了4倍。
六、效果数据：wrk压测对比
同一台机器，同一份数据，60秒压测结果：
| 方案 | QPS | P50 (ms) | P99 (ms) |
|---|---|---|---|
| Flask + lazy load + Jinja2 | 412 | 420 | 1800 |
| FastAPI + selectinload + JSON | 1800 | 85 | 320 |
| FastAPI + selectinload + Redis + gzip | 3200 | 45 | 87 |
从1800ms降到87ms，提升了20倍。CPU使用率从99%降到65%，内存稳定在1.2GB。数据库连接数从峰值200降到50，因为不再有N+1查询。
七、总结：性能优化的优先级
我的经验是：先修架构，再调参数，最后加硬件。这次优化最大的收益来自selectinload（消除N+1）和去掉Jinja2（消除CPU瓶颈），Redis只是锦上添花。如果你也遇到类似问题，建议按这个顺序排查：
- 用PySpy抓线上栈，看CPU/IO热点
- 用
EXPLAIN ANALYZE检查SQL执行计划，重点看Seq Scan和Nested Loop - 对热点数据做缓存，但一定要设TTL并考虑一致性
- 最后才是调连接池、加索引等细粒度优化
不要迷信框架迁移——Flask换FastAPI只提升了10%的QPS，真正的收益来自消除N+1和模板渲染。工具只是放大器，设计才是决定性能的上限。
