Цель/введение
Реактивные паттерны программирования становятся всё более востребованы при реализации высоконагруженных сервисов. Реактивные фреймворки предоставляют инструменты, позволяющие с минимальными затратами на кодирование использовать механизмы асинхронности и многопоточности.
В качестве примера, предлагаю рассмотреть реализацию сервис индексации данных в ElasticSearch. Данные хранятся в MongoDB, ключевые атрибуты которых синхронизируются с ElasticSearch (функционально похоже на Logstash). В проекте используется стек: Java/Spring Boot/Reactor/WebFlux/WebClient/RabbitMQ/MongoDB. На выбор RabbitMQ и MongoDB повлияло, в том числе, наличие реактивных драйверов.
Описание задачи
Сервис должен принимать поток данных из очереди, выбирать связанные данные из базы и передавать их ElasticSearch. Формат данных очереди: действие (index/delete); id документа; имя индекса; тип индекса (опционально).
Через web-интерфейс должен быть реализован функционал добавления, удаления и перестроения индекса.
Должна быть возможность формирования агрегированных полей, содержащих данные из нескольких исходных полей документов, и добавление данных в индекс из связанных коллекций.
Описание индексируемых данных должно быть в формате JSON.
DFD-диаграмма процесса индексации
Схема процесса индексации запросов, поступающих из очереди, выглядит следующим образом:
Алгоритм перестроения индекса выглядит практически также, за исключением того, что в нем отсутствует обработка запросов, отложенных из-за ошибок.
Описание функционала
Описание функционала коснется только работы реактивной части сервиса. Конфигурационные настройки, обработка формата описания индексируемых данных, формирование данных для запросов к ElasticSearch вынесены за рамки данной статьи, но вы можете посмотреть код на GitHub, по ссылке.
Теперь попробуем реализовать эту схему сквозным потоком Reactor, не используя подписки на отдельные элементы, в том числе отправку через WebClient HTTP-запросов и обработку полученных ответов. Отдадим, почти полностью, синхронизацию выполнения Reactor.
Код, запускающий процесс переиндексации выглядит следующим образом:
Task task = new Task(mongoElasticIndex); ParallelFlux dataEventsFlux = reactorRepositoryMongoDB .findAll(mongoElasticIndex.getCollection(), mongoElasticIndex.getProjection()) .parallel(appConfig.getIndexParallelism()) .runOn(Schedulers.boundedElastic()); Flux<Tuple2<String,Document>> processingData = processingData(dataEventsFlux, (p) -> "index", (p) -> (Document)p, (p) -> mongoElasticIndex, Flux.just(), task); task.setDispose(subscribe(processedData, task));
Получаем поток данных из коллекции, настраиваем параллелизм, формируем объект обработки потока и подписываемся на поток. Здесь класс Task – внутренний класс, назначение которого: собирать статистику и предоставлять информацию о выполняемых задачах индексации.
Метод processingData возвращает поток запросов и ответов, отправленных WebClient’ом:
private <T> Flux<Tuple2<String,Document>> processingData(ParallelFlux<T> events, Function<T, String> getAction, Function<T, Document> getDocument, Function<T, MongoElasticIndex> getMongoElasticIndex, Flux<String> mergeFlux, Task task) { return events // Добавление данных к исходному документу из присоединяемых коллекций .transform(joinData(getDocument, getMongoElasticIndex)) // Генерация данных для передачи в ElasticSearch .transform(document2ElasticJson(getAction, getDocument, getMongoElasticIndex)) .sequential() // Агрегирование данных для _bulk .transform(grouping(task)) // Добавление потока данных, на которые не получен ответ от ElasticSearch .mergeWith(mergeFlux) // Отправка запросов в ElasticSearch .transform(postBulk(task)) .subscribeOn(Schedulers.single()) .doOnNext(testAliveResponses(task)) .doOnSubscribe(p-> p.request(appConfig.getMaxSizeBuffer() * 2)) .doOnComplete(() -> { logger.info("Start: {} End: {} read {} write {}", formatDate(task.getStartDate()), formatDate(new Date()), task.getDocumentsRead(), task.getIndexesWrite(), getMaxProcessingRequest()); fileStorage.writeCollection2Files(waitingForResponse); removeTask(task); }); }
Методом transform Reactor соединяем отдельные обработчики потоков. Здесь есть одно существенное ограничение: входящий и исходящий потоки должны быть однотипными (Flux или ParallelFlux). Нельзя, например, с помощью transform встроить обработчик у которого вход Flux, а выход ParallelFlux.
В метод subscribe сервиса инкапсулирована подписка на поток. Ниже приведена его реализация:
private Disposable subscribe(Flux<Tuple2<String,Document>> events, Task task) { return events .subscribe( p -> { if(isNull(task.getMongoElasticIndex())) { // Если задача не переиндексация waitingForResponse.remove(p.getT1()); } int count = Optional.ofNullable(p.getT2().get("items", List.class)) .map(List::size) .orElse(0); task.addIndexesWrite(count); }, e -> { if(task != rabbitMQTask)removeTask(task); fileStorage.writeCollection2Files(waitingForResponse); logger.error("Error: {}", e.getMessage()); } ); }
Далее коротко об отдельных функциях обработки потока.
Загрузка документов
Имеются два варианта загрузки:
Для всех документов основной коллекции индекса
ParallelFlux dataEventsFlux = reactorRepositoryMongoDB .findAll(mongoElasticIndex.getCollection(), mongoElasticIndex.getProjection()) .parallel(appConfig.getIndexParallelism()) .runOn(Schedulers.boundedElastic());
Метод findAll возвращает поток для всех документов коллекции. Parallel и runOn настраивают многопоточность для выборки и дальнейшей обработки.
Для единичного запроса, приходящему из очереди
ParallelFlux dataEventsFlux = reactiveQueue.inboundFlux() .parallel(appConfig.getIndexParallelism()) .runOn(Schedulers.boundedElastic()) .map(msg -> { IndexEvent indexEvent = reactiveQueue.msg2IndexEvent(msg); try { return CreateIndexItem(indexEvent); } catch (IllegalObjectIdException | IOException | ConvertDataException e) { logger.error("{} For message: {}", String.join(", ",throwable2ListMessage(e)), new String(msg.getBody(), StandardCharsets.UTF_8)); return new IndexItem(null, null, null); } }) .filter(e -> nonNull(e.getAction())) .flatMap(item -> Flux.zip("delete".equals(item.getAction()) // Для операции удаления создаётся Document, содержащий _id удаляемого документа ? Flux.just(new Document().append("_id", item.getIdDocument().get("_id"))) // Для операции обновления индекса Document загружается из базы данных : reactorRepositoryMongoDB.find( item.getMongoElasticIndex().getCollection(), item.getIdDocument(), item.getMongoElasticIndex().getProjection()), Flux.just(item) ) .map(d -> new EventDocument(d.getT2().getAction(), d.getT1(), d.getT2().getMongoElasticIndex())) );
Метод inboundFlux интерфейса reactiveQueue возвращает поток для очереди. Parallel и runOn идентичны предыдущему варианту. Далее событие преобразуется из JSON в объект IndexEvent, по содержимому которого документ извлекаются из базы, или создаётся объект для удаления документа из ElasticSearch.
Добавление данных к исходному документу из присоединяемых коллекций
private <T> Function<ParallelFlux<T>, ParallelFlux<T>> joinData(Function<T, Document> getDocument, Function<T, MongoElasticIndex> getMongoElasticIndex) { return (ParallelFlux<T> items) -> items.flatMap(p -> { if(getDocument.apply(p).size() == 1) { return Flux.just(p); } return Flux.fromIterable(getMongoElasticIndex.apply((T) p).getJoinConditions(getDocument.apply(p))) .flatMap(it -> Flux.zip(Flux.just(it.getCollection().getJoinedFieldName()), reactorRepositoryMongoDB.find(getMongoElasticIndex.apply((T) p).getCollection(), it.getCondition(), it.getCollection().getProjection()))) .reduce(p, (acc, t) -> { getDocument.apply(acc).put(t.getT1(), t.getT2()); return acc; }); } ); }
Метод joinData возвращает функциональный объект, добавляющий данные к исходному документу из документов присоединяемых коллекций. Использование flatMap и Flux.zip позволяет асинхронно запускать и обрабатывать потоки, в том числе и потоки, создаваемые запросами к базе данных mongodb. Все вопросы, связанные с синхронизацией, берет на себя Reactor.
Генерация JSON для ElasticSearch
private <T> Function<ParallelFlux<T>, ParallelFlux<String>> document2ElasticJson( Function<T, String> getAction, Function<T, Document> getDocument, Function<T, MongoElasticIndex> getMongoElasticIndex) { return (ParallelFlux<T> items) -> items.map(item -> { String elasticSend; try { Document document = getDocument.apply(item); MongoElasticIndex mongoElasticIndex = getMongoElasticIndex.apply(item); elasticSend = "delete".equals(getAction.apply(item)) ? mongoElasticIndex.deleteBuild(document) : mongoElasticIndex.indexBuild(document); } catch (ConvertDataException e) { throw new RuntimeException(e); } catch (JsonProcessingException e) { throw new UncheckedIOException(e); } return elasticSend; }); }
Из полученного документа формируется JSON-объект модификации индекса в ElasticSearch. Контролируемые исключения приходится конвертировать в неконтролируемые.
Агрегирование данных для _bulk-запроса
Function<Flux<String>, Flux<String>> grouping(Task task) { return (Flux<String> source) -> source .bufferTimeout(appConfig.getMaxSizeBuffer(), Duration.ofMillis(appConfig.getMaxDurationBuffer())) .doOnNext(p -> task.addDocumentsRead(p.size())) .map(p -> String.join("\n", p) ); }
Использование _bulk-запроса к ElasticSearch позволяет существенно снизить трафик и повысить производительность индексации. Объединение отправляемых данных несложно сделать при помощи bufferTimeout. Значениями максимального размера буфера и времени ожидания можно найти компромисс между оперативностью обновления данных в ElasticSearch, размером запроса и производительностью.
Отправка запросов ElasticSearch
public Function<Flux<String>, Flux<Tuple2<String, Document>>> postBulk(Task task) { return (Flux<String> source) -> source .flatMap(buffer -> { if(isNull(task.getMongoElasticIndex())) { // Если задача не переиндексация waitingForResponse.add(buffer); } return Flux.zip(Flux.just(buffer), webClientElastic.post() .uri("/_bulk") .body(BodyInserters.fromValue(buffer)) .retrieve() .onStatus(httpStatus -> httpStatus.equals(HttpStatus.TOO_MANY_REQUESTS), response -> Mono.error(new HttpServiceException("System is overloaded", response.rawStatusCode()))) .onStatus(httpStatus -> httpStatus.is4xxClientError() && !httpStatus.equals(HttpStatus.TOO_MANY_REQUESTS), response -> Mono.error(new RuntimeException("API not found"))) .onStatus(HttpStatus::is5xxServerError, response -> Mono.error(new HttpServiceException("Server is not responding", response.rawStatusCode()))) .bodyToFlux(Document.class) .retryWhen(Retry.backoff(appConfig.getWebClientRetryMaxAttempts(), Duration.ofSeconds(appConfig.getWebClientRetryMinBackoff())) .filter(throwable -> throwable instanceof HttpServiceException) .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { throw new HttpServiceException("External Service failed to process after max retries", HttpStatus.SERVICE_UNAVAILABLE.value()); })) ); }); }
Создаётся поток, отправляющий через WebClient запросы к ElasticSearch. Поток, формируемый методом post WebClient’а, Flux.zip объединяет с запросом, это позволяет при обработке ответа связать полученный ответ с отправленным запросом. С помощью retryWhen, Retry.backoff настроена обработка некоторых ошибок.
Обработка ответов ElasticSearch
private Disposable subscribe(Flux<Tuple2<String,Document>> events, Task task) { return events .subscribe( p -> { if(isNull(task.getMongoElasticIndex())) { // Если задача не переиндексация waitingForResponse.remove(p.getT1()); } int count = Optional.ofNullable(p.getT2().get("items", List.class)) .map(List::size) .orElse(0); task.addIndexesWrite(count); }, e -> { if(task != rabbitMQTask)removeTask(task); fileStorage.writeCollection2Files(waitingForResponse); logger.error("Error: {}", e.getMessage()); } ); }
Обработка ответов ElasticSearch минимальна. Если ответ получен на контролируемый запрос (не запрос на переиндексацию), то запрос удаляется из множества запросов, для которых контролируется получение ответа. Ответы на переиндексацию не контролируются. В полученном ответе атрибут items должен быть списком, содержащим информацию об обработанных документах. На количество элементов в списке увеличивается счетчик обработанных документов.
Настройка WebClient
Основная часть настройки делается в конфигурационном классе, бин возвращает объект WebClient.Builder:
@Bean @Qualifier("elastic") public WebClient.Builder webClientWithTimeout() { final TcpClient tcpClient = TcpClient .create() .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout) .doOnConnected(connection -> { connection.addHandlerLast(new ReadTimeoutHandler(timeout, TimeUnit.MILLISECONDS)); connection.addHandlerLast(new WriteTimeoutHandler(timeout, TimeUnit.MILLISECONDS)); }); return WebClient.builder() .baseUrl(baseUrl +":" + port.toString()) .filter(basicAuthentication(user, password)) .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE); }
В конструкторе сервиса добавляются фильтры, вызываемые при отправке запроса и получении ответа:
this.webClientElastic = webClientElastic .filter(onRequest()) .filter(onResponse()) .build();
Методы, возвращающие фильтры:
private ExchangeFilterFunction onRequest() { return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> { addSendRequest(); int sleepCycleCount = 0; while (getProcessingRequest() > getMaxProcessingRequest()) { try { logger.info("Sleep: {} ProcessingRequest reached {} (MaxProcessingRequest {})", getSleepOverRequest(), getProcessingRequest() - 1, getMaxProcessingRequest()); sleep(getSleepOverRequest()); if (sleepCycleCount++ > appConfig.getSleepCycleCountMax()) { break; } } catch (InterruptedException e) { e.printStackTrace(); } } logger.info("Request: {} {}", clientRequest.method(), clientRequest.url()); return Mono.just(clientRequest); }); } private ExchangeFilterFunction onResponse() { return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> { addReceiveResponse(); logger.info("Response Status {}", clientResponse.statusCode()); return Mono.just(clientResponse); }); }
Фильтры выводят информацию об отправке запросов, получении ответов и модифицируют счетчики отправленных запросов и полученных ответов. Перед отправкой запроса, если превышено количество не полученных ответов, процесс “засыпает” на некоторое время.
Настройка среды выполнения
Для того чтобы запустить этот сервис нам нужны: rabbitmq, mongodb и elasticsearch. Всё это проще установить в Docker. Ещё в самом начале проекта установил Docker Desktop и настроил контейнеры для запуска нужных cервисов. Как это делается можно посмотреть, например, в этой статье. По аналогии установил rabbitmq, mongodb. Добавил конфигурационные файлы и внес изменения в файл docker-compose.yml. Получившиеся настройки Docker можно найти в папке проекта docker-elk. Ниже скриншот запущенного контейнера:
Запуск сервиса
Для тестирования загрузил в базу mongodb 1000 документов. Из Postman и отправляю запрос:
В полученном логе видно, что обработка выполняется в разных потоках:
2022-11-02 15:23:17.396 INFO 8336 --- [ Thread-6] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.418 INFO 8336 --- [ Thread-42] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.447 INFO 8336 --- [ Thread-5] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.462 INFO 8336 --- [ Thread-7] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.475 INFO 8336 --- [ctor-http-nio-1] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.477 INFO 8336 --- [ Thread-33] org.mongodb.driver.connection : Opened connection [connectionId{localValue:855, serverValue:83}] to localhost:27017 2022-11-02 15:23:17.484 INFO 8336 --- [ctor-http-nio-2] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.485 INFO 8336 --- [ Thread-5] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.489 INFO 8336 --- [ctor-http-nio-3] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.493 INFO 8336 --- [ Thread-4] org.mongodb.driver.connection : Opened connection [connectionId{localValue:856, serverValue:84}] to localhost:27017 2022-11-02 15:23:17.566 INFO 8336 --- [ Thread-6] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.588 INFO 8336 --- [ Thread-7] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.622 INFO 8336 --- [ Thread-4] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.651 INFO 8336 --- [ctor-http-nio-4] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.651 INFO 8336 --- [ctor-http-nio-1] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.658 INFO 8336 --- [ctor-http-nio-2] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.658 INFO 8336 --- [ctor-http-nio-3] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.673 INFO 8336 --- [ Thread-52] org.mongodb.driver.connection : Opened connection [connectionId{localValue:940, serverValue:85}] to localhost:27017 2022-11-02 15:23:17.676 INFO 8336 --- [ Thread-7] org.mongodb.driver.connection : Opened connection [connectionId{localValue:941, serverValue:86}] to localhost:27017 2022-11-02 15:23:17.715 INFO 8336 --- [ctor-http-nio-4] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.738 INFO 8336 --- [ Thread-5] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.748 INFO 8336 --- [ Thread-34] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.772 INFO 8336 --- [ Thread-7] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.790 INFO 8336 --- [ Thread-34] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.811 INFO 8336 --- [ctor-http-nio-1] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.818 INFO 8336 --- [ctor-http-nio-2] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.849 INFO 8336 --- [ Thread-36] org.mongodb.driver.connection : Opened connection [connectionId{localValue:993, serverValue:87}] to localhost:27017 2022-11-02 15:23:17.851 INFO 8336 --- [ Thread-36] org.mongodb.driver.connection : Opened connection [connectionId{localValue:994, serverValue:88}] to localhost:27017 2022-11-02 15:23:17.895 INFO 8336 --- [ctor-http-nio-3] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.898 INFO 8336 --- [ctor-http-nio-4] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:17.911 INFO 8336 --- [ Thread-6] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.926 INFO 8336 --- [ Thread-36] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.944 INFO 8336 --- [ Thread-36] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.966 INFO 8336 --- [ Thread-36] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:17.993 INFO 8336 --- [ Thread-36] org.mongodb.driver.connection : Opened connection [connectionId{localValue:1078, serverValue:90}] to localhost:27017 2022-11-02 15:23:18.002 INFO 8336 --- [ Thread-48] org.mongodb.driver.connection : Opened connection [connectionId{localValue:1079, serverValue:89}] to localhost:27017 2022-11-02 15:23:18.041 INFO 8336 --- [ Thread-34] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:18.044 INFO 8336 --- [ctor-http-nio-4] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.044 INFO 8336 --- [ctor-http-nio-2] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.044 INFO 8336 --- [ctor-http-nio-3] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.044 INFO 8336 --- [ctor-http-nio-1] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.059 INFO 8336 --- [ Thread-31] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:18.076 INFO 8336 --- [ Thread-7] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:18.083 INFO 8336 --- [ctor-http-nio-1] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.096 INFO 8336 --- [ctor-http-nio-2] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.135 INFO 8336 --- [ Thread-34] org.mongodb.driver.connection : Opened connection [connectionId{localValue:1156, serverValue:92}] to localhost:27017 2022-11-02 15:23:18.138 INFO 8336 --- [ Thread-39] org.mongodb.driver.connection : Opened connection [connectionId{localValue:1155, serverValue:91}] to localhost:27017 2022-11-02 15:23:18.140 INFO 8336 --- [ Thread-39] ru.mvz.elasticsearch.service.Indexer : Request: POST http://localhost:9200/_bulk 2022-11-02 15:23:18.180 INFO 8336 --- [ctor-http-nio-3] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.180 INFO 8336 --- [ctor-http-nio-4] ru.mvz.elasticsearch.service.Indexer : Response Status 200 OK 2022-11-02 15:23:18.181 INFO 8336 --- [ctor-http-nio-4] ru.mvz.elasticsearch.service.Indexer : Start: 2022-11-02 15:23:17.250 End: 2022-11-02 15:23:18.181 read 1000 write 1000
Теперь проверим, что загрузилось.
Запрос к ElasticSearch показывает наличие индекса с 1000 документами:
И попробуем найти что-то в ElasticSearch:
Получен ответ ElasticSearch с найденным документом!
Заключение
В этом материале мне хотелось привести пример сервиса, реализованного с использованием Spring Boot, WebFlux, WebClient, Reactor - надеюсь, что у меня это получилось.
Несколько выводов:
Реактивные фреймворки, в том числе и Reactor, делает за нас существенную часть работы по реализации асинхронных многопоточных алгоритмов, позволяя сосредоточиться на предметной области.
С их помощью можно, достаточно просто, создавать высоконагруженные сервисы.
Для получения максимального эффекта от перехода на реактивные паттерны программирование нужно чтобы вся цепочка вычислений была реактивной, начиная с драйверов доступа к базам данных, очередям, файлам и т.д.
Ещё раз, репозиторий с кодом и настройками находится здесь
Несколько ссылок на используемые материалы:
Шпаргалка по Spring Boot WebClient
Reactive Programming: Reactor и Spring WebFlux — часть 2
Реактивное программирование со Spring, часть 2 Project Reactor