本文详解如何在 Spring WebFlux + R2DBC 环境下高效实现 PostgreSQL 批量插入，解决 .map(connection -> {...}) 逻辑被跳过、批量执行无响应等常见 Reactive 链中断问题，并提供安全、高性能的批处理实现方案。
本文详解如何在 spring webflux + r2dbc 环境下高效实现 postgresql 批量插入，解决 `.map(connection -> {...})` 逻辑被跳过、批量执行无响应等常见 reactive 链中断问题，并提供安全、高性能的批处理实现方案。
在基于 R2DBC 的响应式数据持久化场景中，直接使用 Mono.from(connectionFactory.create()).map(...) 处理批量插入极易因缺少下游订阅者导致整个操作“静默跳过”——这正是你遇到 .map(connection -> {...}) 未执行的根本原因：R2DBC 的 Batch.execute() 返回的是 Publisher
正确做法是：将 Batch.execute() 显式转为 Flux 并参与响应式链。推荐使用 flatMapMany 桥接连接与批量执行，并确保最终链路可被驱动：
public Mono<Void> articleDestinyBatchStream(List<ArticleDestiny> articles) {
return Mono.from(connectionFactory.create())
.flatMapMany(connection -> {
Batch batch = connection.createBatch();
for (ArticleDestiny a : articles) {
// ⚠️ 重要：必须使用参数化查询，禁止字符串拼接！
batch.add("INSERT INTO article_destiny (country_code, art_number, name, price) " +
"VALUES ($1, $2, $3, $4)")
.bind(0, a.getCountryCode())
.bind(1, a.getArtNumber())
.bind(2, a.getName())
.bind(3, a.getPrice());
}
return Flux.from(batch.execute()); // ← 关键：转为 Flux 并返回
})
.then(); // 完成所有 INSERT 后发出空完成信号
}✅ 关键修正点说明：
- 使用 flatMapMany 而非 map：flatMapMany 可扁平化 Publisher
，使每个 Row 成为流元素，真正触发执行；
- 严禁 SQL 字符串插值（如 '${article.name}'）：不仅存在严重 SQL 注入风险，R2DBC 也不支持 ${} 占位符语法；必须使用 $1, $2 等位置参数 + .bind()；
- then() 是合适的终止操作：它忽略中间结果，只在全部 INSERT 完成后发出 onComplete，契合“批量写入完毕”的语义。
此外，为支撑百万级数据迁移，还需补充以下最佳实践：
-
连接池配置（在 application.yml 中）：
spring: r2dbc: url: r2dbc:postgresql://localhost:5677/article_flux_db username: admin password: secret pool: max-size: 32 min-idle: 8 acquire-timeout: 30s max-life-time: 15m 批量大小调优：客户端 buffer(backPressure) 建议设为 1000–5000（避免单批过大 OOM 或过小降低吞吐），并通过压测确定最优值；
事务控制（可选）：若需强一致性，可在 flatMapMany 内手动开启事务，但注意 R2DBC 批量本身已具备较好原子性，通常无需外层事务；
-
错误处理增强：
.onErrorResume(e -> { log.error("Batch insert failed for {} records", articles.size(), e); return Mono.error(new RuntimeException("Batch persist failed", e)); })
最后，请移除 @SpringBootApplication(exclude = {...}) 中对 R2dbcAutoConfiguration 的排除——除非你有特殊定制需求，否则应交由 Spring Boot 自动装配连接池与模板，确保生产级健壮性。完成上述改造后，原本耗时 20+ 分钟的单条插入，可稳定压缩至 2–5 分钟内，吞吐提升 10 倍以上。
