在 Spring WebFlux webclient 中设置超时

Posted 2023-02-26

【中文标题】在 Spring WebFlux webclient 中设置超时

【英文标题】：set timeout in Spring WebFlux webclient

【发布时间】：2018-08-06 04:11:07

【问题描述】：

我正在使用 Spring Webflux WebClient 从我的 Spring 启动应用程序进行 REST 调用。并且每次都会在 30 秒内超时。

这是我尝试在 Spring webfulx 的 WebClient 中设置套接字超时的一些代码。

 - ReactorClientHttpConnector connector = new ReactorClientHttpConnector(options -> options
           .option(ChannelOption.SO_TIMEOUT, 600000).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 600000));
 - ReactorClientHttpConnector connector = new ReactorClientHttpConnector(
           options -> options.afterChannelInit(chan -> 
                chan.pipeline().addLast(new ReadTimeoutHandler(600000));
            ));
 - ReactorClientHttpConnector connector1 = new ReactorClientHttpConnector(options -> options
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 600000).afterNettyContextInit(ctx -> 
                ctx.addHandlerLast(new ReadTimeoutHandler(600000, TimeUnit.MILLISECONDS));
            ));

并尝试使用“clientConnector”方法将上述连接器设置添加到“WebClient”中。

并且还尝试如下设置超时：

webClient.get().uri(builder -> builder.path("/result/name/sets")
                    .queryParam("q", "kind:RECORDS")
                    .queryParam("offset", offset)
                    .queryParam("limit", RECORD_COUNT_LIMIT)
                    .build(name))
            .header(HttpHeaders.AUTHORIZATION, accessToken)
            .exchange().timeout(Duration.ofMillis(600000))
            .flatMap(response -> handleResponse(response, name, offset));

以上选项都不适合我。

我正在使用 org.springframework.boot:spring-boot-gradle-plugin:2.0.0.M7，其中包含 org.springframework:spring-webflux:5.0.2.RELEASE 的依赖关系。

请在这里提出建议，如果我在这里做错了什么，请告诉我。

【问题讨论】：

已经试过spring-5-webflux-how-to-set-a-timeout-on-webclient

为什么需要 10 分钟的超时时间？

这只是为了测试目的。如果我能够设置超时，那么我将根据我的应用程序对这个时间进行基准测试。

【参考方案1】：

我已尝试重现此问题，但无法重现。使用 reactor-netty 0.7.5.RELEASE。

我不确定你说的是哪个超时。

可以使用ChannelOption.CONNECT_TIMEOUT_MILLIS 配置连接超时。我在“正在连接”日志消息和实际错误之间有 10 秒：

WebClient webClient = WebClient.builder()
    .clientConnector(new ReactorClientHttpConnector(options -> options
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)))
    .build();

webClient.get().uri("http://10.0.0.1/resource").exchange()
    .doOnSubscribe(subscription -> logger.info("connecting"))
    .then()
    .doOnError(err -> logger.severe(err.getMessage()))
    .block();

如果你在谈论读/写超时，那么你可以看看 Netty 的 ReadTimeoutHandler 和 WriteTimeoutHandler。

一个完整的例子可能如下所示：

ReactorClientHttpConnector connector = new ReactorClientHttpConnector(options ->
        options.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10)
                .onChannelInit(channel -> 
                        channel.pipeline().addLast(new ReadTimeoutHandler(10))
                                .addLast(new WriteTimeoutHandler(10));
                return true;
        ).build());

从 Reactor Netty 0.8 和 Spring Framework 5.1 开始，配置现在如下所示：

TcpClient tcpClient = TcpClient.create()
                 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
                 .doOnConnected(connection ->
                         connection.addHandlerLast(new ReadTimeoutHandler(10))
                                   .addHandlerLast(new WriteTimeoutHandler(10)));
WebClient webClient = WebClient.builder()
    .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))
    .build();

也许将以下内容添加到您的 application.properties 将提供有关 HTTP 级别正在发生的事情的更多信息：

logging.level.reactor.ipc.netty.channel.ContextHandler=debug
logging.level.reactor.ipc.netty.http.client.HttpClient=debug

【讨论】：

我们无法设置套接字时间（ChannelOption.SO_TIMEOUT, 600000），甚至无法使用 ReadTimeoutHandler。我的 REST 请求响应时间超过 30 秒，所以我需要设置套接字超时

我的回答没有提到那个选项，是吗？

requestTimeout 确实等于 ChannelOption.SO_TIMEOUT。 connectTimeoutMillis 绝对是关于连接的。

SO_TIMEOUT 用于阻塞 Netty 中的 IO，所以这里不适用。我已经用更多信息编辑了我的答案。

@BrianClozel 感谢您分享适用于 Reactor Netty 0.8 和 Spring Framework 5.1 的设置读取超时和连接超时的更新选项

【参考方案2】：

由于HttpClient.from(tcpClient) 现在在最新的 netty 中已弃用（v0.9.x 并将在 v1.1.0 中删除）。您可以使用 responseTimeout() 并忽略您在其他代码中看到的太多 HTTP 连接配置，并且此实现适用于旧的和新的。

创建 HttpClient

HttpClient httpClient = HttpClient.create().responseTimeout(Duration.ofMillis(500)); // 500 -> timeout in millis

使用 webClient builder fxn .clientConnector() 将 httpClient 添加到 webclient

WebClient
.builder()
.baseUrl("http://myawesomeurl.com")
.clientConnector(new ReactorClientHttpConnector(httpClient))
.build();

此外，网站上提供的大多数实现都没有被弃用，以确保您没有使用已弃用的实现，您可以查看link。

仅供参考：对于较旧的 netty 版本（即 responseTimeout() 在底层使用 tcpConfiguration()，在新版本中已弃用。但是 responseTimeout() 使用 >=v0.9.11 中的新实现，因此即使您将来更改项目的 netty 版本，您的代码也不会中断。

注意：如果你使用的是旧的 netty 版本，它有时默认带有 spring，你可能也可以使用 Brian 的实现。 （虽然不确定）

如果您想了解有关 responseTimeout() 及其工作原理的更多信息，可以查看源代码 here 和 here 或 github gist here。

【讨论】：

我相信这从 v0.9.11 开始可用，而不是所有 v0.9.x。我必须升级使用 v0.9.4 的 Spring 版本。

是的，从 0.9.11 开始提供 responseTimeout。感谢您指出这一点@KaiserCoaster

【参考方案3】：

@im_bhatman 提到的每个 HttpClient.from(TcpClient) 方法已被弃用，这里有一种使用旧方法的方法。

// works for Spring Boot 2.4.0, 2.4.1, and 2.4.2
// doesn't work for Spring Boot 2.3.6, 2.3.7, and 2.3.8
HttpClient httpClient = HttpClient.create()
        //.responseTimeout(Duration.ofSeconds(READ_TIMEOUT_SECONDS))
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIME_MILLIS)
        .doOnConnected(c -> 
                c.addHandlerLast(new ReadTimeoutHandler(READ_TIMEOUT_SECONDS))
                 .addHandlerLast(new WriteTimeoutHandler(WRITE_TIMEOUT_SECONDS));
        );
ClientConnector clientConnector = new ReactorClientHttpConnector(httpClient);
WebClient webClient = WebClient.builder()
        .clientConnector(clientConnector)
        ...
        .build();

我不确定两者的区别

HttpClient#responseTimeout(...) HttpClient#doOnConnected(c -> c.addHandler(new ReadTimeoutHandler(...)))

【参考方案4】：

参考下面的代码块设置超时并使用 webclient 重试

.retrieve()
            .onStatus(
                   (HttpStatus::isError), // or the code that you want
                    (it -> handleError(it.statusCode().getReasonPhrase())) //handling error request
           )
            .bodyToMono(String.class)
            .timeout(Duration.ofSeconds(5))
            .retryWhen(
                    Retry.backoff(retryCount, Duration.ofSeconds(5))
                            .filter(throwable -> throwable instanceof TimeoutException)
           )