Spring WebFlux+R2DBC项目遇连接被Peer关闭异常求助

我正在开发一个Spring WebFlux + R2DBC项目，大部分代码运行正常，但处理部分元素后会收到警告：

r2dbc.mssql.client.ReactorNettyClient : Connection has been closed by peer

随后抛出异常：

ReactorNettyClient$MssqlConnectionClosedException: Connection unexpectedly closed

程序随即停止从R2DBC数据源的Flux中读取数据。

主流水线代码如下：

Sinks.Empty<Void> completionSink = Sinks.empty();

Flux<Event> events = service.getPairs(
                taskProperties.A,
                taskProperties.B);

events
    .flatMap(some operation)
    .doOnComplete(() -> {
        log.info("Finished Job");
        completionSink.emitEmpty(Sinks.EmitFailureHandler.FAIL_FAST);
    })
    .subscribe();
completionSink.asMono().block();

运行时，flatMap默认请求256个元素，之后尝试每次请求1个下一个信号。错误会在处理第280到320个元素之间出现，且非幂等，有时处理280个，有时303、315个等。

我怀疑是网络问题，但不确定具体原因，是否需要调整连接池或其他配置？已尝试调整flatMap请求大小为无界、添加调度器、使用默认R2DBC池，但仍无头绪，恳请帮助。

1. 排查数据库端连接超时配置 #

MSSQL默认存在连接超时/空闲断开的配置，若Flux处理速度较慢，R2DBC连接长时间处于空闲状态（比如flatMap内操作耗时，导致数据库判定连接闲置过久），就会被主动断开：

• 执行sp_configure 'remote query timeout'检查远程查询超时值，默认600秒，若单元素处理耗时接近或超过该值，需调大。
• 检查数据库服务器的TCP连接超时设置（操作系统或SQL Server网络配置层面）。

2. 调整R2DBC连接池的存活与验证参数 #

即使使用默认连接池，也需显式配置连接的存活时间、验证机制，避免复用已被数据库断开的连接：

# application.yml 示例配置
spring:
  r2dbc:
    pool:
      max-idle-time: 30m  # 连接最大空闲时间，需小于数据库的空闲断开时间
      validation-query: SELECT 1  # 从池获取连接前验证有效性
      max-life-time: 1h  # 连接最大存活时间，防止长期占用的连接被断开

验证查询会在每次取连接前执行，确保连接可用，避免拿到已被peer关闭的连接。

3. 优化Flux流水线的背压与连接使用 #

若flatMap内的操作是异步IO或阻塞任务，可能导致R2DBC连接长时间被占用或闲置：

• 若some operation是阻塞操作，必须通过subscribeOn(Schedulers.boundedElastic())将其放到专用线程池，避免阻塞ReactorNetty的IO线程，导致连接处理超时。
• 调整flatMap的并发度（比如flatMap(..., 16)），平衡处理速度与连接占用压力，避免一次性占用过多连接。
• 确保service.getPairs()返回的Flux是按需从数据库拉取数据，而非一次性加载大量数据到内存——若为后者，连接会在数据加载完成后闲置，易被数据库断开。

4. 排查网络层面的闲置连接断开问题 #

若上述配置调整后仍出现问题，需确认网络链路中的中间设备（防火墙、负载均衡）是否会主动断开长时间空闲的TCP连接：

• 检查防火墙的TCP连接超时配置，确保其空闲超时时间大于数据库和连接池的超时设置。
• 启用R2DBC的DEBUG日志，追踪连接的创建、使用、释放过程，定位连接断开的时机：

logging:
  level:
    io.r2dbc.mssql: DEBUG
    reactor.netty: DEBUG

5. 添加连接断开后的重试逻辑 #

网络抖动可能偶发导致连接断开，可为Flux添加针对性的重试机制，确保流水线能恢复：

events
    .flatMap(some operation)
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .filter(throwable -> throwable instanceof MssqlConnectionClosedException))
    .doOnComplete(() -> {
        log.info("Finished Job");
        completionSink.emitEmpty(Sinks.EmitFailureHandler.FAIL_FAST);
    })
    .subscribe();

仅针对连接断开的异常做重试，避免无差别重试引发其他问题。

内容的提问来源于stack exchange，提问作者Barış Vural