Блог

Я использую HTTP-клиент WebFlux из реактивного стека Spring 5 для доступа к внешней службе REST. Я хочу обрабатывать ответы на основе статуса HTTP:

1. Если статус равен 2xx, я хочу вернуться Mono с десертизированным телом ответа.

2. Если статус равен 404, я хочу удалить тело ответа и немедленно вернуть пустой Mono .

3. Для любого другого статуса я хочу удалить тело ответа и вернуть ошибку Mono с. MyBusinessException

Мой код выглядит так:

 webClient.get()
    .uri("/search")
    .syncBody(request)
    .exchange()
    .flatMap { response ->
        when {
            response.statusCode().is2xxSuccessful -> response.bodyToMono(MyResponse::class.java)
            response.statusCode() == NOT_FOUND -> Mono.empty()
            else -> MyBusinessException().toMono<MyResponse>()
        }
     }
  

Я не хочу тратить время на получение и обработку тела ответа, когда оно мне не нужно. JavaDoc для exchange() состояний метода

Вы всегда должны использовать один из методов body или entity ответа, чтобы убедиться, что ресурсы освобождены.

Как именно я должен это сделать, если я хочу удалить тело ответа и немедленно вернуть результат?

Удаление ответа необходимо для того, чтобы соединение можно было повторно использовать для будущих запросов (т.Е. http keep-alive / постоянные соединения).

Чтобы вернуть пустое значение, Mono которое завершается после удаления тела (игнорируя ошибки):

 // Using WebFlux >= 5.2
response.releaseBody()
    // ignore errors
    .onErrorResume(exception -> Mono.empty());


// Using WebFlux < 5.2
response.body(BodyExtractors.toDataBuffers())
    // release DataBuffers
    .doOnNext(DataBufferUtils::release)
    // ignore errors
    .onErrorResume(exception -> Mono.empty())
    // return an empty Mono
    .then();
  

Вернуть пустое Mono значение, которое завершается немедленно, и асинхронно удалить тело (игнорируя ошибки) в фоновом режиме:

 // Using WebFlux >= 5.2
Mono.<Void>empty()
    .doOnSubscribe(s ->
        // drain the body
        response.releaseBody()
            // initiate drain on a separate Scheduler
            .subscribeOn(Schedulers.parallel())
            // subscribe, and ignore errors
            .subscribe())

// Using WebFlux < 5.2
Mono.<Void>empty()
    .doOnSubscribe(s ->
        // drain the body
        response.body(BodyExtractors.toDataBuffers())
            // release DataBuffers
            .doOnNext(DataBufferUtils::release)
            // initiate drain on a separate Scheduler
            .subscribeOn(Schedulers.parallel())
            // subscribe, and ignore errors
            .subscribe())
  

Я бы все же рекомендовал первый вариант, поскольку он немедленно освобождает ресурсы, и, вероятно, это то, что WebClient разработчики имели в виду при его написании и документировании его использования.

Я никогда не использовал второй вариант в производственной системе, поэтому проведите собственное тестирование, чтобы убедиться, что объединение пулов HTTP-соединений работает так, как требуется. При использовании reactor-netty вы можете включить вход в систему отладки reactor.netty.resources.PooledConnectionProvider , чтобы сравнить два подхода.