R2DBC Postgres连接事务空闲异常及批量操作问题求助

问题背景 #

基于Spring Boot 2.7.6、WebFlux和R2DBC-Postgres的应用，在使用flatMap()执行批量数据库插入时出现以下问题：

• 当处理元素数量超过256（Queues.SMALL_BUFFER_SIZE默认值）时，触发TransientDataAccessResourceException，错误信息：

  Cannot exchange messages because the request queue limit is exceeded; nested exception is io.r2dbc.postgresql.client.ReactorNettyClient$RequestQueueException
  

• 异常发生后无"Releasing R2DBC Connection"日志，PostgreSQL连接/会话处于idle in transaction状态，最终导致连接池耗尽，应用无法正常运行。
• 对比现象：使用concatMap()或元素数量≤256时无异常，连接正常释放；修改Queues.SMALL_BUFFER_SIZE或flatMap()并发数至255可临时解决，但非理想方案。

问题代码示例 #

@Transactional
public Mono<Void> insertDummyFooBars() {
        return Flux.fromIterable(IntStream.rangeClosed(1, 260).boxed().collect(Collectors.toList()))
                .log()
                .flatMap(i -> this.repository.save(FooBar.builder().foo("test-" + i).build()))
                .log()
                .concatMap(i -> this.repository.findAll())
                .then();
    }

核心原因 #

1. 请求队列超限：R2DBC-Postgres客户端默认请求队列大小为256，flatMap()默认并发数等于Queues.SMALL_BUFFER_SIZE（256），当并发请求数超过队列上限时触发异常。
2. 连接泄漏：异常发生时，Reactive事务上下文未正确处理连接释放，导致连接绑定在未完成的事务中，处于idle in transaction状态。

解决方案 #

1. 优先使用批量操作API（推荐） #

用R2DBC原生的批量插入方法saveAll()替代并发单条插入，既避免队列超限，又提升执行效率：

@Transactional
public Mono<Void> insertDummyFooBars() {
    List<FooBar> fooBars = IntStream.rangeClosed(1, 260)
            .mapToObj(i -> FooBar.builder().foo("test-" + i).build())
            .collect(Collectors.toList());
    return repository.saveAll(fooBars)
            .log()
            .concatMap(i -> repository.findAll())
            .then();
}

2. 显式控制flatMap()并发数 #

如果必须使用flatMap()，显式设置并发数，确保不超过R2DBC连接池大小和客户端请求队列上限（建议设置为连接池最大大小，默认连接池maxSize为10）：

@Transactional
public Mono<Void> insertDummyFooBars() {
    return Flux.fromIterable(IntStream.rangeClosed(1, 260).boxed().collect(Collectors.toList()))
            .log()
            .flatMap(i -> this.repository.save(FooBar.builder().foo("test-" + i).build()), 10) // 显式设置并发数
            .log()
            .concatMap(i -> this.repository.findAll())
            .then();
}

3. 调整R2DBC客户端请求队列大小 #

通过配置增大R2DBC-Postgres客户端的请求队列上限，适用于确实需要高并发数据库操作的场景（注意避免设置过大导致内存压力）：

spring:
  r2dbc:
    properties:
      max-queue-size: 512 # 调整请求队列大小

4. 确保异常时事务正确回滚 #

Reactive事务会自动在流触发onError时回滚，但需确保错误被正确传播，避免被吞噬。可通过onErrorResume确保事务上下文清理：

@Transactional
public Mono<Void> insertDummyFooBars() {
    return Flux.fromIterable(IntStream.rangeClosed(1, 260).boxed().collect(Collectors.toList()))
            .log()
            .flatMap(i -> this.repository.save(FooBar.builder().foo("test-" + i).build()))
            .log()
            .concatMap(i -> this.repository.findAll())
            .then()
            .onErrorResume(e -> {
                // 确保事务回滚，连接释放
                return Mono.error(e);
            });
}

全局方案建议 #

• 统一规范批量数据库操作：优先使用saveAll()、deleteAll()等批量API，避免不必要的并发单条操作。
• 全局配置连接池与并发数：设置spring.r2dbc.pool.max-size为合理值，同时要求所有flatMap()数据库操作显式指定并发数不超过该值。
• 监控连接状态：通过PostgreSQL的pg_stat_activity视图监控idle in transaction连接，及时发现泄漏问题。

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