如何在 Spring 5 WebFlux WebClient 中设置超时

Posted 2023-02-16

【中文标题】如何在 Spring 5 WebFlux WebClient 中设置超时

【英文标题】：How to set a timeout in Spring 5 WebFlux WebClient

【发布时间】：2018-02-24 09:45:19

【问题描述】：

我正在尝试在我的 WebClient 上设置超时，这是当前代码：

SslContext sslContext = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();

ClientHttpConnector httpConnector = new ReactorClientHttpConnector(opt -> 
    opt.sslContext(sslContext);
    HttpClientOptions option = HttpClientOptions.builder().build();
    opt.from(option);
);
return WebClient.builder().clientConnector(httpConnector).defaultHeader("Authorization", xxxx)
                .baseUrl(this.opusConfig.getBaseURL()).build();

我需要添加超时和池策略，我正在考虑类似的事情：

PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
cm.setMaxTotal(this.applicationConfig.getHttpClientMaxPoolSize());
cm.setDefaultMaxPerRoute(this.applicationConfig.getHttpClientMaxPoolSize());
cm.closeIdleConnections(this.applicationConfig.getServerIdleTimeout(), TimeUnit.MILLISECONDS);

RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(this.applicationConfig.getHttpClientSocketTimeout())
        .setConnectTimeout(this.applicationConfig.getHttpClientConnectTimeout())
        .setConnectionRequestTimeout(this.applicationConfig.getHttpClientRequestTimeout()).build();

CloseableHttpClient httpClient = HttpClients.custom().setDefaultRequestConfig(requestConfig).setConnectionManager(cm).build();

但我不知道如何在我的网络客户端中设置 httpClient

【参考方案1】：

要设置读取和连接超时，我使用下面的方法，因为 SO_TIMEOUT 选项不适用于使用 NIO 的通道（并发出警告 Unknown channel option 'SO_TIMEOUT' for channel '[id: 0xa716fcb2]'）

ReactorClientHttpConnector connector = new ReactorClientHttpConnector(
          options -> options.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
                            .compression(true)
                            .afterNettyContextInit(ctx -> 
                                ctx.addHandlerLast(new ReadTimeoutHandler(5000, TimeUnit.MILLISECONDS));
                            ));
return WebClient.builder()
                .clientConnector(connector)
                .build();

【讨论】：

赞成。这应该是正确的答案。参考文献***.com/a/22897119/839733

太棒了！但是 ReactorClientHttpConnector 在 WebFlux 5.1 中发生了变化。

【参考方案2】：

ReactorClientHttpConnector API 在版本Spring WebFlux 5.1 中更改。

所以我执行以下操作（Kotlin 语法，基于 @joshiste 示例）：

val tcpClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10_000)
    .doOnConnected  connection ->
        connection.addHandlerLast(ReadTimeoutHandler(10))
            .addHandlerLast(WriteTimeoutHandler(10))
    

val myWebClient = webClientBuilder
    .clientConnector(ReactorClientHttpConnector(HttpClient.from(tcpClient)))
    .baseUrl(myEndPoint)
    .build()

2021 年更新

HttpClient.from 在 Reactive Netty 的最新版本中已弃用。它正在复制 tcpClient 的配置。现在我们可以直接配置httpClient了。

val httpClient = HttpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10_000)
    .doOnConnected  connection ->
        connection.addHandlerLast(ReadTimeoutHandler(10))
            .addHandlerLast(WriteTimeoutHandler(10))
    

val myWebClient = webClientBuilder
    .clientConnector(ReactorClientHttpConnector(httpClient))
    .baseUrl(myEndPoint)
    .build()

【讨论】：

使用您的解决方案时，我面对the method from(TcpClient) from the type HttpClient is deprecated。似乎from(TcpClient) 也已被弃用。

【参考方案3】：

随着 Spring Webflux 的更新，这里有一个适用于 Java 的解决方案（基于 Kotlin 的 answer）：

TcpClient timeoutClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS*1000)
    .doOnConnected(
        c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
              .addHandlerLast(new WriteTimeoutHandler(SECONDS)));
return webClientBuilder.baseUrl(YOUR_URL)
       .clientConnector(new ReactorClientHttpConnector(HttpClient.from(timeoutClient)))
       .build();

2021 年更新

由于 HttpClient.from(TcpClient) 已被弃用，现在变得更加容易：

return WebClient.builder()
                .baseUrl(YOUR_URL)
                .clientConnector(new ReactorClientHttpConnector(HttpClient.create()
                                                                          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS * 1000)
                                                                          .doOnConnected(c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
                                                                                               .addHandlerLast(new WriteTimeoutHandler(SECONDS)))))
                .build();

【讨论】：

很有帮助，不过 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS*10) 我认为 10 应该是 1000。

使用您的解决方案时，我面对the method from(TcpClient) from the type HttpClient is deprecated。似乎from(TcpClient) 也被弃用了

@AbhinandanMadaan 修正

【参考方案4】：

WebFlux WebClient 不使用 Apache Commons HTTP 客户端。尽管您可以通过自定义ClientHttpConnector 实现一种解决方案。现有的ReactorClientHttpConnector 是基于 Netty 的。所以，考虑使用 Netty 选项来配置客户端，例如：

ReactorClientHttpConnector connector =
            new ReactorClientHttpConnector(options ->
                    options.option(ChannelOption.SO_TIMEOUT, this.applicationConfig.getHttpClientConnectTimeout()));

或

.onChannelInit(channel -> channel.config().setConnectTimeoutMillis(this.applicationConfig.getHttpClientConnectTimeout()))

更新

我们也可以使用ReadTimeoutHandler:

.onChannelInit(channel -> 
        channel.pipeline()
           .addLast(new ReadTimeoutHandler(this.applicationConfig.getHttpClientConnectTimeout())))

【讨论】：

这似乎是我正在寻找的，只是一个快速的问题是连接超时或请求超时。知道如何设置连接池大小吗？感谢您的帮助

reactor.ipc.netty.options.ClientOptions.Builder中有poolResources()requestTimeout确实等于ChannelOption.SO_TIMEOUT。 connectTimeoutMillis 绝对是关于连接的。

是的，我看到了 poolResources()，我不得不承认我不知道如何使用它：/有什么想法吗？

我试过了：options.poolResources(PoolResources.fixed("myPool", this.applicationConfig.getHttpClientMaxPoolSize()));这是正确的方法吗？

api好像变了，下面是它现在的工作方式***.com/a/53781016/3993662

【参考方案5】：

使用 Spring Webflux 5.1.8 在执行多个使用 WebClient 的后续测试时，我遇到了使用来自 mcoolive 的答案产生以下错误消息的问题。

强制关闭其注册任务未被接受的频道 事件循环 未能提交监听通知任务。事件循环关闭？

添加连接提供者和循环资源解决了我的问题：

final ConnectionProvider theTcpClientPool = ConnectionProvider.elastic("tcp-client-pool");
final LoopResources theTcpClientLoopResources = LoopResources.create("tcp-client-loop");

final TcpClient theTcpClient = TcpClient
    .create(theTcpClientPool)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
    .runOn(theTcpClientLoopResources)
    .doOnConnected(theConnection -> 
        theConnection.addHandlerLast(new ReadTimeoutHandler(mTimeoutInMillisec, TimeUnit.MILLISECONDS));
        theConnection.addHandlerLast(new WriteTimeoutHandler(mTimeoutInMillisec, TimeUnit.MILLISECONDS));
    );

WebClient theWebClient = WebClient.builder()
    .baseUrl(mVfwsServerBaseUrl)
    .clientConnector(new ReactorClientHttpConnector(HttpClient.from(theTcpClient)))
    .build();

【讨论】：

使用您的解决方案时，我面对the method from(TcpClient) from the type HttpClient is deprecated。似乎from(TcpClient) 也被弃用了

我建议查看已弃用方法的文档 (JavaDoc)。有一个关于如何替换不推荐使用的 from() 方法的示例。

【参考方案6】：

我是这样做的（感谢@Artem）

SslContext sslContext = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();

        ClientHttpConnector httpConnector = new ReactorClientHttpConnector(options -> 
            options.sslContext(sslContext);
            options.option(ChannelOption.SO_TIMEOUT, this.applicationConfig.getHttpClientRequestTimeout());
            options.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.applicationConfig.getHttpClientConnectTimeout());
            options.poolResources(PoolResources.fixed("myPool", this.applicationConfig.getHttpClientMaxPoolSize()));
        );

        return WebClient.builder().clientConnector(httpConnector).defaultHeader("Authorization", "xxxx")
                .baseUrl(this.config.getBaseURL()).build();

【讨论】：

这不适用于 Spring 5.1，ReactorClientHttpConnector 不允许再设置选项！

【参考方案7】：

根据上面的评论，如果你想添加一个 Socket Timeout，只需将它作为另一个选项添加到同一个 timeoutClient 中。

TcpClient timeoutClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS*10) //Connect Timeout
    .option(ChannelOption.SO_TIMEOUT,1000) // Socket Timeout
    .doOnConnected(
        c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
              .addHandlerLast(new WriteTimeoutHandler(SECONDS)));
return webClientBuilder.baseUrl(YOUR_URL)
       .clientConnector(new ReactorClientHttpConnector(HttpClient.from(timeoutClient)))
       .build();

【讨论】：

您是否建议同时设置SO_TIMEOUT 和添加ReadTimeoutHandler？

使用您的解决方案时，我面对the method from(TcpClient) from the type HttpClient is deprecated。似乎from(TcpClient) 也被弃用了

【参考方案8】：

您可以使用在 Mono 对象上接受超时的重载 block() 方法。 或者 Mono 对象上有一个 timeout() 方法直接可用。

WebClient webClient = WebClient.builder()
              .baseUrl( "https://test.com" )
              .defaultHeader( HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE )
              .build(); 

webClient.post()
              .uri( "/services/some/uri" )
              .body( Mono.just( YourEntityClassObject ), YourEntityClass.class )
              .retrieve()
              .bodyToMono( String.class )
              .timeout(Duration.ofMillis( 5000 )) // option 1
              .block(Duration.ofMillis( 5000 )); // option 2

【参考方案9】：

您可以提供自定义ReactorNettyHttpClientMapper，而不是创建自己的WebClient.Builder，它将应用于默认WebClient.Builder：

@Configuration
class MyAppConfiguration 

    @Bean
    fun reactorNettyHttpClientMapper(): ReactorNettyHttpClientMapper 
        return ReactorNettyHttpClientMapper  httpClient ->
            httpClient.tcpConfiguration  tcpClient ->
                tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000)
                    .doOnConnected  connection ->
                        connection.addHandlerLast(ReadTimeoutHandler(60))
                            .addHandlerLast(WriteTimeoutHandler(60))
                    
            
        
    

【参考方案10】：

Kotlin 语法！！

webClient
.post()
.body(BodyInserters.fromObject(body))
.headers(headersSetter)
.retrieve()
.bodyToMono<SomeClass>()
.timeout(Duration.ofSeconds(30)) /// specify timeout duration
.doOnNext 
    logger.info "log something"

.onErrorMap  throwable ->
    logger.error"do some transformation"
    throwable