Spring WebClient使用exchangeToFlux()未发起HTTP请求问题排查求助

问题根因 #

• 你用到的exchangeToFlux()返回的是一个Flux类型的冷发布器：Reactor 中所有冷发布器都只有在被订阅后才会触发实际的执行逻辑，你当前的代码只声明了操作链，没有做订阅操作，所以HTTP请求根本不会发起，回调函数自然也不会执行。
• 额外注意：你当前在exchangeToFlux的回调函数中直接调用blockLast()属于错误用法，会阻塞WebClient底层的Netty事件循环线程，轻则导致性能下降，重则触发线程死锁。

修复方案 #

你可以直接对exchangeToFlux返回的Flux做订阅触发请求，同时移除回调内部的阻塞调用，调整后的代码如下：

HttpClient httpClient = HttpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
        .responseTimeout(Duration.ofMillis(5000))
        .doOnConnected(conn ->
                conn.addHandlerLast(new ReadTimeoutHandler(5000, TimeUnit.MILLISECONDS))
                        .addHandlerLast(new WriteTimeoutHandler(5000, TimeUnit.MILLISECONDS)));

WebClient webClient = WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(httpClient))
        .build();
WebClient.RequestHeadersSpec<?> headersSpec = webClient
        .get()
        .uri("http://localhost:8080/v1/users")
        .header(CONTENT_TYPE, "text/csv");

MyResponse<T> myResponse = new MyResponse<>();
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

// 声明操作链，返回最终的Flux
Flux<DataBuffer> resultFlux = headersSpec.exchangeToFlux(clientResponse -> {
    myResponse.setStatus(clientResponse.rawStatusCode());
    myResponse.setContentType(clientResponse.headers().contentType().orElse(null));

    if (clientResponse.statusCode() == HttpStatus.OK) {
        return clientResponse.bodyToFlux(DataBuffer.class);
    }
    return Flux.error(new RuntimeException("请求失败，状态码：" + clientResponse.rawStatusCode()));
});

// 触发订阅+写入输出流，同时阻塞等待操作完成
DataBufferUtils.write(resultFlux, outputStream)
        .blockLast(Duration.of(20, ChronoUnit.SECONDS));

return myResponse;

代码说明 #

• 把原来写在回调内部的DataBufferUtils.write和阻塞逻辑移到了外部，对exchangeToFlux返回的Flux直接做操作和阻塞，既触发了请求执行，也避免了在事件循环线程内做阻塞操作。
• 不需要额外使用CountDownLatch做同步，blockLast已经可以等待整个流处理完成后再返回，完全满足你先拿到响应头、再处理响应体的需求。

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