6

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

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

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

4

5 回答 5

3

我设法配置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

于 2019-07-24T22:14:22.917 回答
1

通过将 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());
于 2019-02-20T18:56:26.180 回答
0

使用 TCP 客户端在 reactor-netty 0.9.x 中执行此操作的最简单方法是使用以下方法,我是从@Vladimir-L 引用的链接中获得的。为您的问题配置“maxIdleTime”。

TcpClient timeoutClient = TcpClient.create(ConnectionProvider.fixed(onnectionPoolName, maxConnections, acquireTimeout,maxIdleTime));
于 2020-04-01T20:11:07.433 回答
0

我目前在 reactor-netty 0.8.2 因为 spring-boot-starter-webflux 并面临同样的问题,连接池在连接完成后保持连接打开 60 秒。

使用这种方法,您无法配置超时,但您可以禁用它:

WebClient.builder()
    .clientConnector(new ReactorClientHttpConnector(
        HttpClient.from(TcpClient.create()).keepAlive(false)))
    .build()
    .get()
    .uri("someurl")
    .retrieve()
    .bodyToMono(String.class)
于 2020-06-04T14:47:44.253 回答
0

对于 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);
于 2021-06-29T11:56:43.873 回答