Postgres R2dbc批量插入Flux流数据异常排查求助

核心原因及排查步骤 #

1. Reactive流未被订阅触发
   Reactive流（如Flux）属于冷流，只有被订阅（subscribe()）才会启动整个执行链。检查你的调用链末尾：

• 如果articleDestinyBatchStream返回的Flux没有被订阅，包括.map在内的所有下游逻辑都不会执行
• 常见错误：仅调用webClient.get().retrieve().bodyToFlux(Article.class).buffer(1000).flatMap(this::articleDestinyBatchStream)，但未添加.subscribe()，也未将其作为Spring Web的返回值（由框架自动订阅）

2. R2dbc连接上下文丢失
   手动处理Connection时，若未在Reactive上下文正确调度，会导致逻辑无法执行：

• 避免手动管理Connection，改用DatabaseClient的批量API，它会自动维护连接生命周期
• 错误示例：直接从ConnectionFactory获取连接，但未绑定到Reactive调度上下文

3. Buffer触发条件未满足
   buffer(n)需要上游流累积到指定数量才会触发下游处理，若上游数据推送慢，会导致逻辑迟迟不执行：

• 改用bufferTimeout(n, Duration.ofSeconds(1))，既按数量拆分，也设置超时兜底，避免无限等待
• 检查上游WebClient返回的Flux是否正常推送数据，可添加.doOnNext(article -> log.debug("收到数据: {}", article.getId()))验证

4. 异常静默吞掉流
   流中发生异常但未被捕获时，会导致流静默终止，看起来像是.map被跳过：

• 在流末尾添加.doOnError(e -> log.error("批量插入失败", e))和.onErrorContinue((e, obj) -> log.warn("处理数据{}失败", obj, e))排查异常
• 开启R2dbc DEBUG级日志，查看SQL执行的详细报错信息

修正后的示例代码 #

// 基于DatabaseClient实现批量插入，避免手动管理连接
private Flux<Void> articleDestinyBatchStream(List<Article> articles) {
    return databaseClient.inConnectionMany(connection -> {
        String sql = "INSERT INTO article_destiny (id, title, content, create_time) VALUES ($1, $2, $3, $4)";
        return Flux.fromIterable(articles)
                .flatMap(article -> connection.createStatement(sql)
                        .bind(0, article.getId())
                        .bind(1, article.getTitle())
                        .bind(2, article.getContent())
                        .bind(3, article.getCreateTime())
                        .add())
                .then(connection.execute())
                .then();
    });
}

// 完整调用链示例
webClient.get()
        .uri("/api/articles/origin")
        .retrieve()
        .bodyToFlux(Article.class)
        .bufferTimeout(1000, Duration.ofSeconds(2)) // 按1000条或2秒拆分批次
        .flatMap(this::articleDestinyBatchStream, 5) // 控制并发数，避免连接耗尽
        .doOnError(e -> log.error("数据迁移失败", e))
        .subscribe(); // 必须订阅触发流执行

性能优化建议 #

• 批量操作前关闭自动提交：connection.setAutoCommit(false)，完成后执行connection.commit()，减少事务开销
• 改用Postgres的COPY命令替代批量INSERT，R2dbc可通过CopyIn实现，性能比普通INSERT高数倍
• 调整R2dbc连接池大小和Postgres的max_connections参数，避免连接耗尽影响批量处理

内容的提问来源于stack exchange，提问作者artsgard