PostgreSQL R2DBC 批量插入性能优化实战指南

本文详解如何在 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<Row>（即 Flux<Row>），而你用 .map() 包裹后未对其进行订阅或转换为有效流，Reactor 会直接丢弃该 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>，使每个 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 倍以上。