WebClient报Reactor Netty连接提前关闭问题排查求助

在使用WebClient获取数据并处理时，随机出现上述错误。待处理的分页数据量较大，后续需执行大量阻塞式数据库更新与插入操作（未实现响应式JDBC）。

核心处理方法如下：

private Mono<String> migrateSomeData() {
    return getAllPages()
        .flatMapIterable(Page::getItems)
        .filter(this::isValidItem)
        .doOnNext(item -> doSomeLogging())
        .filter(item-> this.checkIfAlreadyProcessed(item))
        .flatMap(this::mapToDto)
        .flatMap(this::persist)
        .doOnNext(this::saveAsProcessed)
        .collectList()
        .map(this::getTotalAmountOfProcessed);
}

注：getAllPages方法通过Flux.expand递归获取分页数据，该逻辑已验证正常。

已尝试调整WebClient配置，代码如下：

@Bean
public WebClient someClient(
    ReactiveOAuth2AuthorizedClientManager clientAuthorizedClientManager) {

  ServerOAuth2AuthorizedClientExchangeFilterFunction oauth =
      new ServerOAuth2AuthorizedClientExchangeFilterFunction(clientAuthorizedClientManager);
  oauth.setDefaultClientRegistrationId("clientId");

  final int size = 16 * 1024 * 1024;
  final ExchangeStrategies strategies = buildClientWithExtendedResponseSize(size);

  return WebClient.builder()
      .defaultHeader("Accept", MediaType.APPLICATION_JSON_VALUE)
      .defaultHeader("subscription-key", someKeyValue)
      .clientConnector(createWiretappedClientHttpConnector(this.getClass()))
      .filter(oauth)
      .exchangeStrategies(strategies)
      .build();
}

private static ExchangeStrategies buildClientWithExtendedResponseSize(int size) {
  return ExchangeStrategies.builder()
      .codecs(codecs -> codecs.defaultCodecs().maxInMemorySize(size))
      .build();
}

private ClientHttpConnector createWiretappedClientHttpConnector(Class<?> invokedClass) {
  HttpClient httpClient =
      HttpClient.create()
          .option(ChannelOption.SO_KEEPALIVE, true)
          .responseTimeout(Duration.ofMinutes(5))
          .doOnConnected(
              conn ->
                  conn.addHandlerLast(new ReadTimeoutHandler(5 * 60))
                      .addHandlerLast(new WriteTimeoutHandler(5 * 60)))
          .compress(true)
          .wiretap(
              invokedClass.getCanonicalName(), LogLevel.TRACE, AdvancedByteBufFormat.TEXTUAL);
  return new ReactorClientHttpConnector(httpClient);
}

且已通过本地WireMock模拟API测试，问题仍存在，可排除服务端配置问题。

疑问：是否可能因数据处理过慢，导致WebClient返回的响应未及时被消费，进而触发连接提前关闭？

是的，这完全有可能是数据处理过慢导致的问题。

你的数据流逻辑从getAllPages()获取分页数据后，直接串联了大量阻塞式DB操作（persist、checkIfAlreadyProcessed、saveAsProcessed均为阻塞方法）。在Reactor的线程模型中，WebClient的响应消费默认由Reactor Netty的IO线程处理，如果后续阻塞操作长时间占用IO线程，会导致IO线程无法及时读取响应数据，最终触发连接超时或被提前关闭。

具体原因拆解： #

• getAllPages()通过Flux.expand递归拉取分页，每一页的响应需要被及时消费才能维持连接活性或触发下一页请求；
• 阻塞式DB操作直接在IO线程执行时，会占用线程资源，导致WebClient的响应数据无法被及时处理，服务器或客户端连接池会判定连接超时，主动关闭连接，从而抛出「Connection prematurely closed BEFORE response」错误。

解决方案建议： #

1. 将阻塞操作转移到专用线程池：
   用Mono.fromCallable包装阻塞方法，并通过subscribeOn指定专用线程池，避免占用IO线程：

   // 定义专用DB线程池
   private final ExecutorService dbExecutor = Executors.newFixedThreadPool(10);
   
   // 在数据流中替换阻塞操作调用
   .filter(item -> Mono.fromCallable(() -> checkIfAlreadyProcessed(item))
                       .subscribeOn(Schedulers.fromExecutor(dbExecutor))
                       .block()) // 注意：这里如果用block会破坏响应式，正确做法是改成链式调用
   // 正确的链式写法：
   .flatMap(item -> Mono.fromCallable(() -> checkIfAlreadyProcessed(item))
                       .subscribeOn(Schedulers.fromExecutor(dbExecutor))
                       .filter(Boolean::booleanValue)
                       .map(__ -> item))
   .flatMap(item -> Mono.fromCallable(() -> persist(item))
                       .subscribeOn(Schedulers.fromExecutor(dbExecutor)))
   

2. 控制并发度：
   默认flatMap并发度为256，对阻塞操作来说过高，可指定合理的并发数，避免线程池耗尽：

   .flatMap(item -> Mono.fromCallable(() -> persist(item))
                       .subscribeOn(Schedulers.fromExecutor(dbExecutor)),
           10) // 限制并发数为10
   

3. 优化背压处理：
   针对大数据量分页，使用背压操作符平衡上游拉取速度与下游处理速度，避免缓冲区溢出：

   getAllPages()
       .onBackpressureBuffer(10, bufferOverflow -> log.warn("数据处理过慢，触发背压缓冲"))
       .flatMapIterable(Page::getItems)
       // 后续操作...
   

4. 调整连接池参数：
   进一步优化Reactor Netty连接池配置，避免连接耗尽或等待超时：

   HttpClient.create(ConnectionProvider.builder("custom-db-pool")
           .maxConnections(50)
           .pendingAcquireTimeout(Duration.ofMinutes(2))
           .maxIdleTime(Duration.ofMinutes(10))
           .build())
   

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