JavaNotes

Spring Reactor

3 min read

Spring Reactor

Project Reactor — реализация Reactive Streams, на которой построен Spring WebFlux. Два основных типа:

  • Mono<T> — от нуля до одного элемента;
  • Flux<T> — от нуля до многих элементов.

Сам объект Mono или Flux обычно описывает вычисление. Работа начинается при подписке, поэтому до subscribe или возврата цепочки инфраструктуре запрос может вообще не выполниться.

Сигналы и операторы

Mono<Invoice> invoice =
        repository.findById(id)
                .switchIfEmpty(Mono.error(new InvoiceNotFound(id)))
                .filter(Invoice::isPayable)
                .switchIfEmpty(Mono.error(new InvalidState(id)))
                .flatMap(paymentClient::pay)
                .map(receipt -> Invoice.paid(id, receipt.number()));
  • map преобразует обычное значение T -> R;
  • flatMap связывает асинхронный шаг T -> Publisher<R>;
  • filter может превратить Mono<T> в пустой Mono;
  • switchIfEmpty задаёт поведение для отсутствующего значения;
  • doOn... наблюдает сигнал и не должен менять бизнес-результат.

null не является элементом reactive stream. Отсутствие представляется пустым publisher.

Ошибки

client.load(id)
        .timeout(Duration.ofSeconds(2))
        .retryWhen(Retry.backoff(2, Duration.ofMillis(100))
                .filter(TemporaryNetworkException.class::isInstance))
        .onErrorMap(ClientException.class,
                error -> new InvoiceUnavailable(id, error));

onErrorReturn и onErrorResume превращают ошибку в значение или другую цепочку. Их нельзя добавлять «на всякий случай»: fallback способен скрыть потерю данных. Retry применяют только к повторяемой операции и ограничивают по числу, времени и типу ошибки.

flatMap, порядок и параллельность

flatMap может выполнять несколько внутренних publisher одновременно, поэтому результат способен изменить порядок. concatMap обрабатывает по одному и сохраняет порядок. flatMapSequential допускает конкурентную работу, но выдаёт результаты в исходной последовательности.

Flux.fromIterable(ids)
        .flatMap(client::load, 8) // не более восьми запросов одновременно
        .collectList();

Параметр concurrency ограничивает давление на зависимость. Без лимита быстрая входная последовательность может перегрузить клиентский пул.

Scheduler

publishOn переключает контекст выполнения последующих операторов, subscribeOn влияет на подписку и источник.

Для короткой вынужденной блокировки используют ограниченный boundedElastic, но это переходный мост:

Mono.fromCallable(() -> legacyClient.load(id))
        .subscribeOn(Schedulers.boundedElastic());

Так блокировка не исчезает, а переносится с event loop на отдельный ограниченный пул. Для устойчивой системы важны timeout и предел очереди.

Reactor и сервер

Reactor не является сервером. Spring WebFlux может работать поверх Reactor Netty, а также в поддерживаемом servlet-контейнере. Reactor Netty использует небольшое число event-loop потоков, поэтому блокирующая операция на таком потоке задерживает множество соединений.

Нельзя без необходимости вызывать:

  • block() и blockFirst() в reactive request path;
  • Thread.sleep;
  • блокирующий JDBC;
  • синхронный HTTP-клиент;
  • тяжёлый CPU-код на event loop.

Контекст

ThreadLocal ненадёжен после переключения потоков. Для request-scoped технических данных Reactor предоставляет Context:

Mono.deferContextual(context ->
        audit(context.get("traceId")))
    .contextWrite(context -> context.put("traceId", traceId));

Контекст распространяется снизу вверх относительно места contextWrite; это частая причина неожиданного отсутствия значения.

Тестирование

StepVerifier.create(service.findName(42))
        .expectNext("sample")
        .verifyComplete();

StepVerifier проверяет элементы, terminal-сигнал и виртуальное время. Это точнее, чем вызвать block() и проверить только одно значение.

На собеседовании

Почему не надо вызывать subscribe() в service? Подпиской должен владеть внешний слой: он управляет жизненным циклом, отменой и обработкой ошибок. Service обычно возвращает Mono или Flux.

Когда map, а когда flatMap? map, если функция возвращает обычное значение; flatMap, если она возвращает publisher.

Решает ли WebFlux проблему медленной БД? Нет. Он экономнее использует потоки ожидания, но производительность зависимости, лимиты соединений и backpressure остаются.

Связанные заметки: Реактивное программирование Java, Spring WEB, Http протокол.

Graph View

Backlinks