索引设计
数据库索引是查询性能的第一道防线。eventd v2 的关键索引：
-- 单字段索引（等值查询）
CREATE INDEX idx_orders_user_id ON orders(user_id);
-- 复合索引：等值列在前，范围/排序列在后
CREATE INDEX idx_orders_status_created ON orders(status, created_at DESC);
-- 部分索引（只索引活跃订单，减少索引大小）
CREATE INDEX idx_orders_pending ON orders(created_at)
WHERE status = 'PENDING';
-- 表达式索引（JSONB 字段查询）
CREATE INDEX idx_events_payload_type ON events((payload->>'type'));
索引设计原则：
- WHERE 子句中的字段优先建索引，但低选择性的列单独建索引通常没用
- 复合索引字段顺序：先放等值条件列，再放范围或排序列。
(status, created_at)能同时服务status = ?和ORDER BY created_at；反过来(created_at, status)对这类查询帮助有限。「区分度高的放前面」只在多个等值列之间作为次要参考 - 覆盖索引：查询字段都在索引中，避免回表
- 不要过度索引：写操作（INSERT/UPDATE）需要维护索引
EXPLAIN ANALYZE
PostgreSQL 的查询计划分析：
EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)
SELECT * FROM orders
WHERE user_id = 1 AND status = 'COMPLETED'
ORDER BY created_at DESC
LIMIT 10;
输出解读：
[
{
"Plan": {
"Node Type": "Index Scan",
"Index Name": "idx_orders_user_id",
"Actual Rows": 10,
"Actual Loops": 1,
"Actual Startup Time": 0.031,
"Actual Total Time": 0.052,
"Shared Hit Blocks": 4,
"Shared Read Blocks": 0
},
"Planning Time": 0.084,
"Execution Time": 0.076
}
]
FORMAT JSON 的最外层是数组，指标在 Plan 里；顶层另有 Planning Time、Execution Time。
| 指标 | 含义 | 怎么判断 |
|---|---|---|
Actual Total Time | 该节点累计执行时间（毫秒） | 看占比最大的节点，绝对值要和查询目标比较 |
Actual Startup Time | 返回第一行前的时间 | 与 Total 差距大说明在等数据 |
Shared Hit Blocks | 从共享缓冲池命中的块数 | 高说明走内存，不代表索引一定好 |
Shared Read Blocks | 从磁盘读取的块数 | 高说明发生物理 I/O |
Rows Removed by Filter | 扫描后过滤掉的行数 | 远大于 Actual Rows 说明索引选择差 |
如果看到 Seq Scan（全表扫描）但查询有 WHERE 条件，说明缺少索引或索引未命中。
N+1 问题
N+1 是 ORM/Repository 的常见病：
// 先查 1 次用户列表
userRepo.findByStatus("ACTIVE") // 1 query
.flatMap(user ->
// 每个用户再查订单：N queries
orderRepo.findByUserId(user.id()) // N queries
)
.collectList()
.subscribe();
如果有 100 个活跃用户，会执行 101 次数据库查询。
解决方案 A：JOIN
@Query("""
SELECT u.id AS user_id, u.name, o.id AS order_id, o.symbol
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.status = :status
""")
Flux<UserOrderRow> findUserOrderRows(@Param("status") String status);
// 应用层把扁平行按用户聚合成对象
userRepo.findUserOrderRows("ACTIVE")
.collectMultimap(UserOrderRow::userId)
.subscribe(byUser -> { ... });
一次查询返回所有数据，应用层组装。返回类型必须是扁平行（一行一个用户-订单组合），不能直接声明成 UserWithOrders：Spring Data R2DBC 不会替你把多行折叠进同一个对象的集合字段。缺点是返回数据有冗余（用户字段重复）。
解决方案 B：批量 IN 查询
userRepo.findByStatus("ACTIVE")
.collectList()
.flatMap(users -> {
List<Long> userIds = users.stream().map(User::id).toList();
return orderRepo.findByUserIdIn(userIds)
.collectMultimap(Order::userId)
.map(ordersMap -> users.stream()
.map(u -> new UserWithOrders(u,
List.copyOf(ordersMap.getOrDefault(u.id(), List.of()))))
.toList());
});
先收集所有用户 ID，再执行一次 IN 查询。总共 2 次查询。
解决方案 C：响应式批量（推荐）
userRepo.findByStatus("ACTIVE")
.buffer(50) // 每 50 个用户一批
.flatMap(batch -> {
List<Long> ids = batch.stream().map(User::id).toList();
return orderRepo.findByUserIdIn(ids)
.collectMultimap(Order::userId)
.map(orders -> batch.stream()
.map(u -> new UserWithOrders(u,
List.copyOf(orders.getOrDefault(u.id(), List.of()))))
.toList());
})
.flatMapIterable(list -> list) // List<UserWithOrders> → Flux<UserWithOrders>
.subscribe();
buffer(50) 控制批量大小。PostgreSQL 单条语句最多接受 65535 个绑定参数（协议里参数个数是 16 位），而且 IN 列表过长会让计划膨胀、内存占用上升；分批是同时控制这两件事，不是为了绕开 32767 这个并不存在的上限。
flatMap 并发度与连接池
orderRepo.findByStatus("PENDING")
.flatMap(order -> processOrder(order), 10) // 并发度 10
.subscribe();
并发度 10 意味着同时处理 10 条订单。如果 processOrder 涉及数据库写操作，需要确保连接池有足够连接：
ConnectionPool pool = new ConnectionPool(ConnectionPoolConfiguration.builder(factory)
.maxSize(20) // ≥ flatMap 并发度 + 其他路径的预留
.build());
注意这不是 Reactor Netty 的 ConnectionProvider——那是 HTTP 客户端用的；R2DBC 连接池是 io.r2dbc.pool.ConnectionPool，两者不要混。池大小至少要覆盖 flatMap 并发度，再加其他查询和健康检查的余量；具体设多少取决于数据库能承受的连接数和压测结果，不存在 并发度 × 2 + 10 这种通用公式。
分页：OFFSET vs 游标
OFFSET 分页（简单但不稳定）
PageRequest page = PageRequest.of(100, 20); // index 0-based：第 101 页，每页 20 条
orderRepo.findByStatus("PENDING", page);
生成 LIMIT 20 OFFSET 2000。问题：
- 数据变化时，同一页可能重复或遗漏
- OFFSET 大时性能差（需要扫描前 N 条）
游标分页（稳定高效）
@Query("""
SELECT * FROM orders
WHERE status = :status AND id > :lastId
ORDER BY id
LIMIT :pageSize
""")
Flux<Order> findByStatusWithCursor(
@Param("status") String status,
@Param("lastId") Long lastId,
@Param("pageSize") int pageSize
);
// 使用
Long lastId = 0L;
do {
List<Order> page = orderRepo.findByStatusWithCursor("PENDING", lastId, 20)
.collectList().block();
// 处理 page
lastId = page.isEmpty() ? null : page.get(page.size() - 1).id();
} while (lastId != null);
游标分页使用索引直接定位，O(log n)，不随页数增长而变慢。适合无限滚动、数据导出等场景。上面的 block() 只适合离线批处理；在服务端接口里应把循环写成递归的 Flux.expand 或 Flux.generate，保持整条链非阻塞。
写入优化：批量插入
R2DBC Statement 的 add() 可以批量写入：
Statement stmt = conn.createStatement(
"INSERT INTO events (type, payload) VALUES ($1, $2)");
for (Event event : events) {
stmt.bind("$1", event.type())
.bind("$2", event.payload())
.add();
}
// 每个 add() 对应一个 Result，必须用 Flux.from 消费全部结果
Flux.from(stmt.execute())
.flatMap(Result::getRowsUpdated)
.reduce(0L, Long::sum);
省掉的主要是网络往返：一次发送多条绑定值，而不是每条一次往返。能快多少取决于单条 SQL 的延迟、批量大小和事务边界，必须用自己的数据测；「快 10–50 倍」不是可以照搬的常数。另外 add() 仍是多条独立的 INSERT，不会合并成 PostgreSQL 的 COPY 或单条多值 INSERT。
小结
| 优化手段 | 方案 | 效果 |
|---|---|---|
| 索引 | CREATE INDEX + 复合索引 | 避免全表扫描 |
| 计划分析 | EXPLAIN (ANALYZE) | 定位慢查询原因 |
| N+1 | JOIN / IN 查询 / buffer + flatMap | 减少查询次数 |
| 并发度 | flatMap(fn, concurrency) | 匹配连接池容量 |
| 分页 | 游标（id > :lastId） | 稳定 + O(log n) |
| 批量写入 | Statement.add() | 减少网络往返 |
下一节 8.3 连接健康检查 ：连接池 metrics、健康端点、自动重连与故障检测。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。