如何在 reactor-netty 中配置池连接空闲超时

Question

我正在使用带有连接池的 reactor-netty http 客户端（0.7.X 系列），并且想配置池连接的空闲超时但不知道在哪里。

更准确地说，我需要配置 reactor-netty http 客户端连接池，使其自动关闭在可配置超时内未看到任何活动的连接。这些连接是打开的，但在某些（可配置的）时间内没有字节被传入或传出。

如何配置 reactor-netty http 客户端以抢先关闭空闲连接？

Answer 1

我设法配置WebClient（通过底层）从reactor-netty 0.8.9TcpClient的连接池中删除超时空闲连接我的解决方案部分基于有关IdleStateHandler的官方文档，并扩展了我对如何在创建HttpClient.我是这样做的：public class IdleCleanupHandler extends ChannelDuplexHandler {    @Override    public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception {        if (evt instanceof IdleStateEvent) {            final IdleState state = ((IdleStateEvent) evt).state();            if (state == IdleState.ALL_IDLE) { // or READER_IDLE / WRITER_IDLE                // close idling channel                ctx.close();            }        } else {            super.userEventTriggered(ctx, evt);        }    }}...public static WebClient createWebClient(final String baseUrl, final int idleTimeoutSec) {    final TcpClient tcpClient = TcpClient.create(ConnectionProvider.fixed("fixed-pool"))        .bootstrap(bootstrap -> BootstrapHandlers.updateConfiguration(bootstrap, "idleTimeoutConfig",            (connectionObserver, channel) -> {                channel.pipeline()                    .addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTimeoutSec))                    .addLast("idleCleanupHandler", new IdleCleanupHandler());            }));    return WebClient.builder()        .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))        .baseUrl(baseUrl)        .build();}重要更新：我的进一步测试表明，在bootstrap挂钩期间添加处理程序会破坏池，并且套接字（通道）不会被Connection.添加处理程序的正确方法是：public static WebClient createWebClient(final String baseUrl, final int idleTimeoutSec) {    final TcpClient tcpClient = TcpClient.create(ConnectionProvider.fixed("fixed-pool"))        .doOnConnected(conn -> {            final ChannelPipeline pipeline = conn.channel().pipeline();            if (pipeline.context("idleStateHandler") == null) {                pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTimeoutSec))                        .addLast("idleCleanupHandler", new IdleCleanupHandler());            }        });    return WebClient.builder()        .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))        .baseUrl(baseUrl)        .build();}注意：在reactor-netty0.9.x 中将有一个标准的方法来为连接池中的连接配置空闲超时，请参阅此提交：https ://github.com/reactor/reactor-netty/pull/792

Answer 2

通过将 netty 写入和读取超时处理程序添加到通道管道，我能够在 0.7.x 分支上完成此操作。但是，在 0.8.x 上，这种方法不再有效。HttpClient httpClient = HttpClient    .create((HttpClientOptions.Builder builder) -> builder    .host(endpointUrl.getHost())    .port(endpointUrl.getPort())    .poolResources(PoolResources.fixed(connectionPoolName, maxConnections, timeoutPool))    .afterChannelInit(channel -> {        channel.pipeline()                // The write and read timeouts are serving as generic socket idle state handlers.                .addFirst("write_timeout", new WriteTimeoutHandler(timeoutIdle, TimeUnit.MILLISECONDS))                .addFirst("read_timeout", new ReadTimeoutHandler(timeoutIdle, TimeUnit.MILLISECONDS));    })    .build());

Answer 3

由于 spring-boot-starter-webflux 我目前在 reactor-netty 0.8.2 并面临同样的问题，连接池在连接完成后保持连接打开 60 秒。使用这种方法，您无法配置超时，但您可以禁用它：WebClient.builder()    .clientConnector(new ReactorClientHttpConnector(        HttpClient.from(TcpClient.create()).keepAlive(false)))    .build()    .get()    .uri("someurl")    .retrieve()    .bodyToMono(String.class)

Answer 4

对于 Reactor Netty 版本 1，您需要创建一个reactor.netty.resources.ConnectionProvider包含空闲时间配置的文件，然后在创建reactor.netty.http.client.HttpClient.我正在使用 Spring，所以我使用它来创建一个 Spring org.springframework.http.client.reactive.ClientHttpConnector，如下所示。        ConnectionProvider connectionProvider = ConnectionProvider.builder("Name")                .maxIdleTime(Duration.ofSeconds(10))                .build();        HttpClient httpClient = HttpClient.create(connectionProvider)                .compress(true);        return WebClient.builder()                .clientConnector(new ReactorClientHttpConnector(httpClient))                .baseUrl(host);