一、问题背景
先说清楚场景。这是一个内部商品服务，技术栈是 FastAPI 0.110.0 + SQLAlchemy 2.0.29 + PostgreSQL 15.6，部署在 4C8G 的容器里，uvicorn 起 4 个 worker，前面挂 Nginx。接口 GET /api/v1/products/search 承担商品列表查询，支持按分类、关键词、价格区间过滤，返回分页结果，每个商品还要带上它的标签列表。
上线初期数据量小（大概 3 万商品），一切正常。数据涨到 80 万商品、平均每个商品 4 个标签之后，问题来了：监控上 P99 从 200ms 一路爬到 1.2s，QPS 到 210 左右就开始出现 504。业务方反馈"搜索页转圈转得让人想砸键盘"。
这篇文章就是把这个过程完整复盘一遍。不是那种"我加了个缓存就好了"的爽文，中间踩的坑我都写进去了。
二、环境与版本
先把环境交代清楚，不然数据没法复现：
- Python 3.11.8
- FastAPI 0.110.0 / Starlette 0.36.3
- uvicorn 0.29.0，
--workers 4 - SQLAlchemy 2.0.29（async，asyncpg 0.29.0 驱动）
- PostgreSQL 15.6，shared_buffers 2GB，work_mem 16MB
- Redis 7.2.4（缓存层）
- 压测工具：locust 2.24.1，压测机 8C16G，与被压服务同内网
压测脚本固定 500 并发、持续 3 分钟、随机分类 + 随机价格区间，模拟真实流量分布。
三、方案设计：先定位，再动手
我的原则是：没有 profiling 数据之前不写一行优化代码。很多性能问题看起来像"数据库慢"，实际可能是序列化、可能是事件循环被阻塞、也可能是连接池太小。拍脑袋优化大概率是白干。
排查分三步：
- 进程级 profiling：用 py-spy 采样，看时间到底花在哪个函数上；
- SQL 级分析：把慢查询抓出来，用
EXPLAIN (ANALYZE, BUFFERS)看执行计划； - 针对性优化：N+1 改成批量预加载、补索引、加缓存。
第一步：py-spy 采样
py-spy 的好处是不用改代码、不用重启，直接 attach 到运行中的进程。线上环境我一般用 --nonblocking 避免影响业务：
# 安装
pip install py-spy==0.3.14
# 找到 uvicorn worker 的 pid
ps -ef | grep uvicorn
# 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --nonblocking
# 或者直接看 top 函数（更适合快速判断）
py-spy top --pid 12345 --nonblocking
py-spy top 的输出很直观，我截了关键部分：
GIL: 78.00% Threads: 4
23.0% select (asyncpg/protocol/protocol.pyx)
18.0% _execute (sqlalchemy/engine/base.py)
12.0% serialize (pydantic/main.py)
9.0% get (redis/asyncio/client.py) # 当时已有少量缓存
...
注意，采样里 asyncpg 的 select 占了 23%，说明大量时间在等数据库返回。但这只能说明"数据库交互多"，不能说明"单条查询慢"。结合 QPS 只有 210 这个事实，我高度怀疑是查询次数太多——也就是 N+1。
第二步：抓 SQL
SQLAlchemy 开 echo 太重，我用了事件监听，只记录超过阈值的慢查询：
# app/db_profiler.py
import time
from sqlalchemy import event
from sqlalchemy.engine import Engine
SLOW_THRESHOLD_MS = 20
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
conn.info.setdefault("query_start_time", []).append(time.perf_counter())
@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
start = conn.info["query_start_time"].pop()
elapsed_ms = (time.perf_counter() - start) * 1000
if elapsed_ms > SLOW_THRESHOLD_MS:
# 生产环境请走 logger，不要 print
print(f"[SLOW {elapsed_ms:.1f}ms] {statement[:200]}")
在 main.py 里 import 一下这个模块即可生效。结果抓出来一个请求打了 41 条 SQL：1 条查商品主表，1 条 count，剩下 39 条全是 SELECT ... FROM tags WHERE product_id = $1——典型 N+1，每条 tags 查询 3~8ms，39 条就是 150ms 往上。但 P99 是 1.2s，光 N+1 解释不了，继续往下看。
第三步：EXPLAIN 分析主查询
把商品主表那条查询单独拿出来：
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM products
WHERE category_id = 12
AND price BETWEEN 100 AND 500
AND name ILIKE '%手机%'
ORDER BY created_at DESC
LIMIT 20 OFFSET 0;
执行计划里最刺眼的是：
Seq Scan on products (cost=0.00..48213.00 rows=1 width=312)
(actual time=0.021..412.335 rows=20 loops=1)
Filter: ((category_id = 12) AND (price >= 100) AND (price <= 500) AND (name ~~* '%手机%'))
Rows Removed by Filter: 799980
Buffers: shared hit=1204 read=36112
全表扫描，扫了 80 万行只为了拿 20 行，光这一条 SQL 就 412ms。而且 ILIKE '%手机%' 前缀通配，B-tree 索引根本用不上。到这里问题就清楚了：慢的根因是主查询全表扫描 + 标签 N+1，两者叠加。
四、核心实现
4.1 消除 N+1：selectinload 批量预加载
SQLAlchemy 2.0 里，一对多关系用 selectinload 最稳（它会把所有 product_id 收集起来，用一条 WHERE product_id IN (...) 批量查，而不是 join 导致结果集膨胀）：
# app/repository/product_repo.py
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from app.models import Product
async def search_products(session, category_id: int, price_min: int, price_max: int,
keyword: str | None, offset: int, limit: int):
stmt = (
select(Product)
.where(
Product.category_id == category_id,
Product.price.between(price_min, price_max),
)
.options(selectinload(Product.tags)) # 关键：批量预加载
.order_by(Product.created_at.desc())
.offset(offset)
.limit(limit)
)
if keyword:
stmt = stmt.where(Product.name.ilike(f"%{keyword}%"))
result = await session.execute(stmt)
return result.scalars().unique().all()
改完之后单请求 SQL 数从 41 降到 2（主查询 + 一次 tags IN 查询），这部分省了大概 150ms。但主查询本身还是全表扫描，得从索引下手。
4.2 索引：复合索引 + pg_trgm
针对过滤条件 category_id + price + created_at，建复合索引。这里有个细节：ORDER BY created_at DESC 要能走索引，索引列顺序得把等值条件放前面、排序字段放后面：
-- 复合索引：等值列在前，排序列在后
CREATE INDEX CONCURRENTLY idx_products_cat_price_created
ON products (category_id, price, created_at DESC);
-- 关键词模糊匹配用 pg_trgm，注意 ILIKE 需要 gin_trgm_ops
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX CONCURRENTLY idx_products_name_trgm
ON products USING gin (name gin_trgm_ops);
-- 更新统计信息，让优化器选对计划
ANALYZE products;
CONCURRENTLY 是必须的，线上表直接 CREATE INDEX 会锁表。代价是建索引期间会多扫一遍表，80 万行大概 40 秒，可以接受。
建完索引再 EXPLAIN，同一查询从 412ms 降到 8.7ms，计划变成：
Index Scan using idx_products_cat_price_created on products
(actual time=0.031..7.842 rows=20 loops=1)
Buffers: shared hit=9 read=12
4.3 缓存：Redis 两级策略
到这一步 P99 大概 180ms 了，但压测 QPS 上到 800 之后又开始抖。原因是热门分类的查询被反复打到 DB，PG 的连接池（pool_size=20）扛不住。加缓存。
策略上我做了两个区分：
- 列表结果缓存：key 由查询参数哈希生成，TTL 60s，接受短暂不一致；
- 缓存穿透防护：空结果也缓存（TTL 10s），避免恶意参数打穿。
代码用装饰器实现，注意 async 场景下 Redis 客户端要复用连接池：
# app/cache.py
import hashlib, json
import redis.asyncio as redis
from functools import wraps
pool = redis.ConnectionPool.from_url(
"redis://10.0.0.21:6379/0",
max_connections=50,
decode_responses=True,
)
r = redis.Redis(connection_pool=pool)
def cache_result(ttl: int = 60, empty_ttl: int = 10):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
raw = json.dumps({"a": [str(a) for a in args[1:]], "k": kwargs},
sort_keys=True, ensure_ascii=False)
key = f"cache:{func.__name__}:{hashlib.md5(raw.encode()).hexdigest()}"
cached = await r.get(key)
if cached is not None:
return json.loads(cached)
result = await func(*args, **kwargs)
payload = json.dumps(result, ensure_ascii=False, default=str)
# 空结果短 TTL，防穿透
await r.set(key, payload, ex=empty_ttl if not result else ttl)
return result
return wrapper
return decorator
在 service 层套上：
@cache_result(ttl=60)
async def search_products_cached(session, category_id, price_min, price_max,
keyword, offset, limit):
rows = await search_products(session, category_id, price_min, price_max,
keyword, offset, limit)
return [serialize_product(p) for p in rows]
五、踩坑与优化
坑 1：selectinload 和 unique() 的关系。一开始没加 .unique()，SQLAlchemy 2.0 直接抛 InvalidRequestError。一对多预加载后需要去重，这是 2.0 的强制行为，别省。
坑 2：pg_trgm 索引对中文的坑。gin_trgm_ops 是基于三元组的，中文短词（比如两个字的关键词）三元组切分效果差，实测两字关键词命中率一般。我的处理是：关键词长度 < 3 时走 LIKE '关键词%' 前缀匹配，配合普通 B-tree 索引，效果反而更好。这个分支我写在了 service 层。
坑 3：缓存 key 里带 session 对象。装饰器第一版把 session 一起序列化了，直接报 TypeError: Object of type AsyncSession is not JSON serializable。所以代码里用 args[1:] 跳过 session，只对业务参数做哈希。
坑 4：Redis 连接数打满。压测 500 并发时出现 ConnectionError: Too many connections。原因是每个请求新建 client。改成全局 ConnectionPool（上面代码里的写法）后解决，max_connections=50 对当前量级足够。
坑 5：CREATE INDEX 忘加 CONCURRENTLY。第一次在预发环境直接建，表被锁了 20 秒，接口全部超时。生产环境务必 CONCURRENTLY。
六、效果数据
所有数据都是同一套 locust 脚本、500 并发、3 分钟，取稳定段：
| 阶段 | P50 | P95 | P99 | QPS | 单请求 SQL 数 |
|---|---|---|---|---|---|
| 优化前 | 420ms | 890ms | 1210ms | 210 | 41 |
| 消除 N+1 后 | 180ms | 340ms | 460ms | 480 | 2 |
| 加复合索引后 | 62ms | 130ms | 185ms | 950 | 2 |
| 加 Redis 缓存后 | 21ms | 58ms | 86ms | 2410 | 0.3（命中率 87%） |
缓存命中率 87% 是稳定后的数字，冷启动阶段会低一些。DB 侧 CPU 从峰值 92% 降到 23%，连接池等待基本消失。
七、总结
这次优化最值钱的不是那几个索引和缓存，而是"先 profiling 再动手"这个流程。如果一开始就凭直觉加缓存，N+1 和全表扫描的问题会被缓存掩盖，一旦缓存失效（冷启动、大促缓存击穿），数据库瞬间被打穿，那才是真正的灾难。
几个可以直接抄走的经验：
py-spy top --nonblocking是线上定位性能问题的第一选择，零侵入；- SQLAlchemy 2.0 的 N+1 用
selectinload+.unique()，别用joinedload处理一对多； - 复合索引列顺序：等值条件在前，排序字段在后，范围条件放最后；
- 中文模糊搜索，pg_trgm 不是万能药，短关键词走前缀匹配更实在；
- 缓存装饰器记得跳过 session 之类的不可序列化对象，空结果也要缓存。
如果你们线上也有类似的接口，建议先跑一遍 py-spy，大概率能发现和我一样的问题。有踩到别的坑欢迎评论区交流。
