реактор-нетто: использование HTTP-клиента keep-alive - PullRequest
Новая
       9

реактор-нетто: использование HTTP-клиента keep-alive

0 голосов
/ 29 апреля 2019

Я использую reactor-netty для запроса набора URL. Большинство URL-адресов принадлежат одним и тем же хостам. reactor-netty создает новое TCP-соединение для каждого URL, даже если соединение с хостом уже установлено для предыдущего URL. Некоторые серверы сбрасывают новые соединения или начинают медленно реагировать, когда установлены сотни одновременных соединений.

Образец кода: 

    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
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] Channel cleaned, now 0 active connections and 240 inactive connections

Можно ли запросить несколько URL-адресов на одном хосте, используя HTTP-клиент keep-alive через одно и то же TCP-соединение с хостом? Если нет, как я могу ограничить количество одновременных подключений к одному и тому же хосту или выполнить запросы к одному и тому же хосту последовательно (следующий запрос только после получения ответа на предыдущий)?

Я использую Californium-SR6 поезд выпуска.

1 Ответ

1 голос
/ 05 мая 2019

Да, нетто реактора поддерживает поддержание активности, повторное использование соединения и пул соединений.

Обратите внимание, что .flatMap - это асинхронная операция, которая обрабатывает внутренние потоки параллельно.Поэтому при вызове group.flatMap(... внутренние запросы будут выполняться параллельно.И поскольку они выполняются параллельно, необходимо установить несколько соединений.

Если вы хотите выполнять запросы к одному и тому же хосту последовательно, измените свой пример на использование group.concatMap вместо .flatMap.

Если вы все еще хотите выполнять их параллельно, но ограничить количество активных запросов для отдельного хоста, измените свой пример на использование одной из перегруженных версий .flatMap, которая принимает параметр concurrency.

Кроме того, поскольку вы используете HttpClient.create(), в вашем примере используется глобальный пул http-соединений по умолчанию.Если вы хотите больше контролировать пул соединений, вы можете указать другой ConnectionProvider через HttpClient.create(ConnectionProvider).

...