reactor-netty：使用 keep-alive HTTP 客户端

Question

我reactor-netty用来请求一组 URL。大多数 URL 属于同一主机。reactor-netty即使已经为前一个 URL 建立了与主机的连接，似乎也为每个 URL 建立了一个全新的 TCP 连接。当建立数百个同时连接时，一些服务器会断开新连接或开始缓慢响应。

代码示例：

    Flux.just(...)

    .groupBy(link -> {

        String host = "";

        try {

            host = new URL(link).getHost();

        } catch (MalformedURLException e) {

            LOGGER.warn("Cannot determine host {}", link, e);

        }

        return host;

    })

    .flatMap(group -> {

        HttpClient client = HttpClient.create()

                .keepAlive(true)

                .tcpConfiguration(tcp -> tcp.host(group.key()));

        return group.flatMap(link -> client.get()

            .uri(link)

            .response((resp, cont) -> resp.status().code() == 200 ? cont.aggregate().asString() : Mono.empty())

            .doOnSubscribe(s -> LOGGER.debug("Requesting {}", link))

            .timeout(Duration.ofMinutes(1))

            .doOnError(e -> LOGGER.warn("Cannot get response from {}", link, e))

            .onErrorResume(e -> Flux.empty())

            .collect(Collectors.joining())

            .filter(s -> StringUtils.isNotBlank(s)));

    })

    .blockLast();

在日志中，我看到同一远程主机的本地端口不同，并且活动和非活动连接的总和远高于不同主机的数量。这就是为什么我认为这reactor-netty不是重用已经建立的连接。

DEBUG [2019-04-29 08:15:18,711] reactor-http-nio-10 r.n.r.PooledConnectionProvider: [id: 0xaed18e87, L:/192.168.1.183:56832 - R:capcp2.naad-adna.pelmorex.com/52.242.33.4:80] Releasing channel

DEBUG [2019-04-29 08:15:18,711] reactor-http-nio-10 r.n.r.PooledConnectionProvider: [id: 0xaed18e87, L:/192.168.1.183:56832 - R:capcp2.naad-adna.pelmorex.com/52.242.33.4:80] Channel cleaned, now 1 active connections and 239 inactive connections

...

DEBUG [2019-04-29 08:15:20,158] reactor-http-nio-10 r.n.r.PooledConnectionProvider: [id: 0xd6c6c5db, L:/192.168.1.183:56965 - R:capcp2.naad-adna.pelmorex.com/52.242.33.4:80] Releasing channel

keep-alive是否可以通过与主机的同一 TCP 连接使用 HTTP 客户端在同一主机上请求多个 URL ？如果没有，我如何限制同时连接到同一主机的数量或按顺序执行对同一主机的请求（下一个请求只有在收到对前一个响应的响应后）？

我使用Californium-SR6发布火车。

Answer 1

是的，reactor netty 支持 keep-alive、连接重用和连接池。请注意，这.flatMap是一个并行处理内部流的异步操作。因此，当您调用group.flatMap(...内部请求时，将并行执行。而且由于它们是并行执行的，因此需要建立多个连接。如果您想按顺序执行对同一主机的请求，请将您的示例更改为使用group.concatMap而不是.flatMap.如果您仍想并行执行它们，但限制对单个主机的活动请求的数量，则将您的示例更改为使用.flatMap带concurrency参数的重载版本之一。此外，由于您使用的是HttpClient.create()，因此您的示例使用默认的全局 http 连接池。如果您想对连接池进行更多控制，可以指定不同的ConnectionProvidervia HttpClient.create(ConnectionProvider)。