一、背景：接口慢到被运维盯上
上周四，运维同事在工作群里@我：订单详情接口p99耗时已经飙到1.2秒，网关超时阈值是2秒，再涨下去要出事故了。我看了下监控面板，这个GET /api/v1/orders/{order_id}接口平时调用量不大，但每次调用都要查订单主表、商品快照、用户地址、优惠券明细，代码是半年前写的，当时测试环境数据量小没暴露问题。
生产库订单表50万行，订单明细表200万行，用户表80万行。接口逻辑不复杂：查询订单→遍历details列表→逐条查商品和优惠券。典型的N+1查询，加上FastAPI默认的JSONResponse序列化，直接把RT打上去了。
二、环境与版本
先交代一下压测和优化环境，方便大家复现：
Python: 3.11.9
FastAPI: 0.115.6
Uvicorn: 0.30.6 (workers=4, loop=uvloop)
SQLAlchemy: 2.0.36
asyncpg: 0.30.0
PostgreSQL: 15.4 (work_mem=16MB, shared_buffers=4GB)
Redis: 7.2 (单节点, maxmemory 2GB)
压测工具: wrk 4.2.0 (线程数8, 连接数128, 时长30s)
压测命令：
wrk -t8 -c128 -d30s --latency http://localhost:8000/api/v1/orders/ORD20241101
基线数据（优化前）：
Running 30s test @ http://localhost:8000/api/v1/orders/ORD20241101
8 threads and 128 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 215.32ms 41.78ms 512.91ms 72.11%
Req/Sec 46.21 7.89 68.00 64.20%
Latency Distribution
50% 198.12ms
75% 231.44ms
90% 268.90ms
99% 412.37ms
1386 requests in 30.00s, 0.96MB read
Requests/sec: 46.20
平均215ms，吞吐46 req/s。对于一个查询接口来说，这个数字完全可以接受——但前提是数据量不涨。问题是订单表每个月涨10万行，三个月后必炸。
三、第一步：用cProfile和py-spy定位瓶颈
我习惯先profile再改代码，避免瞎猜。直接在本地跑生产数据副本，用cProfile抓热路径：
import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient
client = TestClient(app)
profiler = cProfile.Profile()
profiler.enable()
for _ in range(50): # 跑50次取平均
client.get("/api/v1/orders/ORD20241101")
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative").print_stats(20)
关键输出：
ncalls tottime percall cumtime percall filename:lineno(function)
100 0.021 0.000 8.342 0.083 order_service.py:87(get_order_detail)
100 0.014 0.000 6.891 0.069 order_service.py:112(_fetch_details)
150 0.008 0.000 4.223 0.028 asyncpg/protocol.py:78(bind_execute)
150 0.006 0.000 3.891 0.026 sqlalchemy/dialects/postgresql/asyncpg.py:345(_execute_context)
100 0.002 0.000 1.231 0.012 main.py:210(serialize_response)
数据很清楚：
_fetch_details累计耗时6.89秒，占总耗时82.6%。这里面是150次独立的SQL查询（1次主订单 + 1次details列表 + 148次商品/优惠券单查）。serialize_response耗时1.23秒，占比14.7%。这是FastAPI默认的JSONResponse用标准库json.dumps，对于嵌套datetime和Decimal对象效率极低。
四、第二步：消灭N+1查询，用selectinload合并且只取必要列
原来的代码长这样（简化版）：
async def get_order_detail(order_id: str, session: AsyncSession):
order = await session.get(Order, order_id)
details = await session.execute(
select(OrderDetail).where(OrderDetail.order_id == order_id)
)
detail_list = details.scalars().all()
result = []
for d in detail_list:
product = await session.get(Product, d.product_id) # 每次一条SQL
coupon = await session.get(Coupon, d.coupon_id) if d.coupon_id else None
result.append(serialize(d, product, coupon))
return result
改成SQLAlchemy 2.0的selectinload，一次性join出所有关联表。同时用with_only_columns只取需要的字段，避免把整个Product对象（包含大字段如description、image_urls）加载到内存：
from sqlalchemy.orm import selectinload
from sqlalchemy import select
async def get_order_detail(order_id: str, session: AsyncSession):
stmt = (
select(Order)
.where(Order.id == order_id)
.options(
selectinload(Order.details).selectinload(OrderDetail.product),
selectinload(Order.details).selectinload(OrderDetail.coupon),
)
)
result = await session.execute(stmt)
order = result.scalar_one_or_none()
if not order:
raise HTTPException(404, "Order not found")
return order
这里有个关键点：SQLAlchemy 2.0的selectinload会生成第二条SQL，用WHERE product_id IN (...)一次性查出所有商品。150次查询变成3次。但注意，selectinload默认会把整行数据都查出来，如果你只需要name和price，可以配合load_only：
from sqlalchemy.orm import load_only
# 在options中加load_only限制字段
.options(
selectinload(Order.details).selectinload(OrderDetail.product)
.load_only(Product.id, Product.name, Product.price),
selectinload(Order.details).selectinload(OrderDetail.coupon)
.load_only(Coupon.id, Coupon.discount, Coupon.expired_at),
)
改完这步先压测一轮：
Thread Stats Avg Stdev Max +/- Stdev
Latency 68.44ms 15.21ms 210.56ms 68.33%
Req/Sec 145.72 18.44 210.00 63.21%
4372 requests in 30.00s, 3.02MB read
Requests/sec: 145.73
平均68ms，吞吐146 req/s。N+1解决了，但还不够快。剩下的大头是序列化。
五、第三步：ORJSON + 自定义响应模型，序列化耗时降80%
FastAPI默认的JSONResponse底层是标准库json.dumps，处理datetime和Decimal时会走默认的str()转换，不仅慢还容易丢精度。换用orjson，它原生支持ISO格式的datetime，并且对Decimal有优化。
安装：
pip install orjson==3.10.7
在FastAPI中全局替换响应类：
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
app = FastAPI(default_response_class=ORJSONResponse)
但光换响应类还不够，ORM对象转dict的serialize函数也要优化。原来的代码用了Pydantic v2的model_dump()，在v2.8.2版本下性能其实不错，但如果你用的是v1或者手写dict，建议改用Pydantic v2的from_attributes=True：
from pydantic import BaseModel, ConfigDict
from datetime import datetime
from decimal import Decimal
class ProductOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: str
name: str
price: Decimal
class OrderDetailOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
product: ProductOut
quantity: int
unit_price: Decimal
class OrderOut(BaseModel):
id: str
created_at: datetime
total_amount: Decimal
details: list[OrderDetailOut]
关键是设置model_config，这样Pydantic可以直接从ORM对象构建，不用手动转dict。再配合ORJSON，序列化耗时从1.23秒降到0.24秒。
第二轮压测：
Thread Stats Avg Stdev Max +/- Stdev
Latency 28.91ms 6.58ms 88.34ms 70.12%
Req/Sec 342.18 31.22 480.00 65.30%
10265 requests in 30.00s, 8.22MB read
Requests/sec: 342.18
平均29ms，吞吐342 req/s。已经比基线好10倍，但还有优化空间。
六、第四步：Redis缓存热数据，压测从29ms到12.8ms
这个接口的订单数据一旦生成就不会变（除非退款，但那是另一个接口），非常适合缓存。我加了Redis缓存，TTL设30分钟，key设计为order:{order_id}，value用ORJSON序列化后的dict。
缓存策略有两个细节：
- 缓存穿透保护：如果订单不存在，在Redis中缓存一个空对象，TTL设30秒，防止恶意请求穿透到DB。
- 缓存击穿：高并发下如果缓存过期，会有大量请求同时打到DB。我用了Redis的
SET NX EX做分布式锁，只让一个请求去查DB并回填缓存，其他请求等待锁释放后直接读缓存。
实现代码：
import orjson
import redis.asyncio as aioredis
from fastapi import Depends
redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=False)
async def get_order_cached(order_id: str, session: AsyncSession):
cache_key = f"order:{order_id}"
cached = await redis_client.get(cache_key)
if cached:
return orjson.loads(cached)
# 分布式锁防止击穿
lock_key = f"lock:{order_id}"
lock_acquired = await redis_client.set(lock_key, "1", nx=True, ex=5)
if not lock_acquired:
# 等待锁释放后重试读缓存
for _ in range(10):
await asyncio.sleep(0.05)
cached = await redis_client.get(cache_key)
if cached:
return orjson.loads(cached)
raise HTTPException(503, "Service busy, try again")
try:
order = await get_order_detail(order_id, session)
# 序列化后存缓存，TTL 1800秒
await redis_client.setex(cache_key, 1800, orjson.dumps(order))
return order
finally:
await redis_client.delete(lock_key)
注意：缓存存储的是序列化后的dict，所以get_order_detail返回的不再是ORM对象，而是dict。需要在service层做转换，或者用Pydantic的model_validate再转一次。我用的是后者，保证类型安全。
第三轮压测：
Thread Stats Avg Stdev Max +/- Stdev
Latency 12.83ms 3.21ms 42.55ms 71.08%
Req/Sec 780.44 58.12 1024.00 62.30%
23413 requests in 30.00s, 18.77MB read
Requests/sec: 780.41
平均12.8ms，吞吐780 req/s。对比基线：延迟降93.9%，吞吐提升16.9倍。
七、踩坑记录与最终总结
优化过程中踩了三个坑，值得记录：
坑1：selectinload与分页冲突。如果订单的details超过100条，selectinload会生成IN查询，但PostgreSQL对IN列表长度有限制（默认65535个参数）。我的场景不会这么多，但如果你做通用封装，建议加个阈值判断，超过50条改用lazy="raise"报错，提醒开发者改代码。
坑2：ORJSON序列化Decimal的精度问题。orjson默认会把Decimal转成float，导致精度丢失。需要在序列化时手动处理：
def default_serializer(obj):
if isinstance(obj, Decimal):
return str(obj) # 转成字符串保留精度
raise TypeError
order_json = orjson.dumps(order, default=default_serializer)
坑3：Redis连接池耗尽。压测时没有限制连接数，导致Redis连接数飙到200个，触发了服务端maxclients限制。解决方案是显式指定连接池大小：
redis_client = aioredis.from_url(
"redis://localhost:6379",
max_connections=50,
socket_connect_timeout=2,
socket_timeout=2
)
最终数据汇总：
| 指标 | 优化前 | N+1修复后 | +ORJSON | +Redis缓存 |
|---|---|---|---|---|
| Avg Latency | 215ms | 68ms | 29ms | 12.8ms |
| P99 Latency | 412ms | 150ms | 60ms | 25ms |
| Throughput | 46 req/s | 146 req/s | 342 req/s | 780 req/s |
| DB Queries per request | 150 | 3 | 3 | 0 (缓存命中) |
总结一下：性能优化80%靠的是消除冗余IO（N+1查询、序列化开销），20%靠缓存。先profile再动手，别上来就加Redis，那是本末倒置。另外，所有优化都要有压测数据支撑，否则你无法判断改动是否有效。
代码已推到公司私有仓库，后续会考虑把缓存策略抽成通用装饰器，支持TTL和锁参数配置，但那是下周的事了。这篇先写到这里，有问题评论区见。
