Spring WebFlux: когда брать, Mono/Flux, R2DBC

← назад к разделу

Большую часть времени обычный сервис ничего не считает — он ждёт: ответа базы, соседнего сервиса, очереди. В Spring MVC каждый запрос получает поток, и пока ответ не пришёл, поток занят ожиданием. WebFlux устроен иначе: потоков мало, ждать им запрещено, а результат приходит событием. За это платят другой моделью программирования — и большая часть ошибок в реактивных сервисах растёт из непонимания этой модели, а не из незнания операторов. С приходом виртуальных потоков в Java 21 сам выбор всё чаще складывается не в пользу WebFlux, поэтому статья начинается с него.

Поток на запрос спит, пока ждёт ответа

Tomcat под Spring MVC держит пул из 200 потоков (server.tomcat.threads.max). Запрос занимает один из них на всё время обработки — включая те 30–300 мс, пока база или соседний сервис готовят ответ. Поток в это время ничего не делает, но и другому запросу его не отдать. Когда все 200 заняты ожиданием, 201-й запрос ложится в очередь приёма, хотя процессор простаивает. Добавить потоков можно, но каждый платформенный поток — мегабайт зарезервированного стека плюс работа планировщика ОС, и на нескольких тысячах потоков переключения стоят дороже самой работы.

WebFlux решает это малым числом потоков. Сервер под ним — Reactor Netty, потоков цикла событий (event loop) у него столько, сколько процессоров, но не меньше четырёх. Обработчик запроса не ждёт: он запускает обращение к базе и возвращает поток циклу, тот берёт следующий запрос. Когда ответ базы придёт, он попадёт в очередь цикла как событие, и обработка продолжится. Четырёх потоков хватает на тысячи одновременных запросов ровно потому, что ни один из них не спит.

Виртуальные потоки Java 21 закрывают ту же проблему с другого конца. Модель остаётся «поток на запрос», но поток дешёвый: его стек живёт в куче, а когда он ждёт ответа от сокета, JVM снимает его с несущего платформенного потока и отдаёт тот другому. В Spring Boot 3.2+ это одна строка, spring.threads.virtual.enabled=true, после которой Tomcat выполняет каждый запрос в виртуальном потоке, а код с JdbcTemplate, @Transactional и ThreadLocal остаётся прежним. Ограничение JDK 21: внутри synchronized виртуальный поток прилипает к несущему и ждёт вместе с ним; в JDK 24 это исправили.

Разница между второй и третьей дорожкой — в цене: виртуальные потоки оставляют привычный код, WebFlux меняет его целиком — типы, порядок выполнения, правила для контекста и ошибок.

Когда WebFlux оправдан, а когда хватит MVC

Первое условие жёсткое: реактивным должен быть весь путь запроса. Драйвер базы — R2DBC, а не JDBC; HTTP-клиент — WebClient; клиенты очереди, кеша и хранилища файлов — тоже неблокирующие. Один JdbcTemplate.queryForObject() в середине цепочки держит поток цикла событий, а их четыре, — преимущество исчезает, сложность остаётся. Половинчатости внутри одного модуля не бывает: если в проекте лежат и spring-boot-starter-web, и spring-boot-starter-webflux, Spring Boot поднимает сервлетный стек с MVC, а контроллеры с Mono работают через асинхронный режим сервлета — неблокирующего сервера вы не получите, пока не уберёте -web или не переключите spring.main.web-application-type=reactive.

WebFlux выигрывает там, где данные идут потоком и где соединений много, а работы на каждое мало. Отдача событий клиенту по SSE или бесконечный Flux из очереди — в Reactor это тот же тип, что и одиночный ответ; в MVC есть SseEmitter, но каждое открытое соединение там стоит потока — с виртуальными дешёвого, зато без обратного давления, о котором ниже. Тысячи открытых WebSocket или долгих опросов (long polling) — в WebFlux соединение не занимает ничего, кроме памяти на буферы. Пять–десять параллельных обращений к другим сервисам в одном запросе с таймаутами и повторами — zip и flatMap выражают это короче, чем CompletableFuture. И команда, уже пишущая на Reactor в соседних сервисах: единый стиль дешевле двух.

WebFlux проигрывает в самом частом случае: обычные операции над записями поверх JPA. JPA блокирующая, замены с ленивой загрузкой и отслеживанием изменений в реактивном мире нет, а нагрузка в сотни и тысячи запросов в секунду на таком сервисе снимается MVC с виртуальными потоками. И проигрывает, когда команда реактивность не знает: ошибки из следующих разделов — забытая подписка, блокирующий вызов, потерянный контекст — в проде стоят дороже сэкономленных потоков.

Если решение всё-таки WebFlux, зависимости такие:

implementation("org.springframework.boot:spring-boot-starter-webflux")
implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")
runtimeOnly("org.postgresql:r2dbc-postgresql")

spring-boot-starter-data-jpa в тот же модуль технически ляжет, и приложение стартует — но каждый вызов репозитория придётся уводить на отдельный пул потоков; как именно и во что это превращается — в разделе про subscribeOn.

Цепочка операторов — это рецепт: до подписки ничего не происходит

Метод возвращает Mono<Order>, и кажется, что заказ уже загружен и просто завёрнут в обёртку. Это не так. Mono<T> — не значение, а описание того, как значение получить: ноль или один элемент, когда-нибудь потом. Flux<T> — то же для последовательности от нуля до бесконечности. Каждый оператор в цепочке — map, flatMap, filter — не выполняет работу, а дописывает шаг в описание. Работа начинается, когда кто-то вызовет subscribe(); до этого момента запрос к базе даже не отправлен. Так устроены «холодные» источники, и в Reactor такие почти все.

В контроллере подписчик — сам WebFlux: вы возвращаете Mono из метода, он подписывается и отдаёт результат в ответ. Отсюда самая частая ошибка: вызвать auditRepo.save(event), выбросить полученный Mono<Void> и удивиться, что в таблице пусто — описание построено, подписки не было, записи нет. Побочный эффект нужно включить в возвращаемую цепочку, .then(auditRepo.save(event)), либо, если ответ не должен его ждать, подписаться явно.

Вторая ловушка того же корня — Mono.just(expensiveCall()). Аргумент just вычисляется прямо здесь, на потоке вызывающего, до всякой подписки — просто потому, что это обычный вызов метода Java. Mono.fromCallable(() -> expensiveCall()) откладывает вызов до подписки, а Mono.defer(() -> ...) делает то же для целой цепочки. Третье следствие: каждая подписка выполняет описание заново, поэтому retry() повторяет запрос к базе целиком.

Обычный контроллер целиком выглядит так:

@RestController
@RequiredArgsConstructor
@RequestMapping("/api/v1/orders")
public class OrderController {

    private final OrderRepository repo;
    private final PricingClient pricing;

    @GetMapping("/{id}")
    public Mono<OrderResponse> get(@PathVariable UUID id) {
        return repo.findById(id)
            .flatMap(order -> pricing.calculate(order)
                .map(price -> OrderResponse.from(order, price)))
            .switchIfEmpty(Mono.error(() -> new OrderNotFoundException(id)));
    }

    @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<OrderEvent> stream() {
        return repo.streamEvents();
    }
}

Mono.error здесь получает не готовое исключение, а функцию: иначе объект исключения со стек-трейсом создавался бы при каждой сборке цепочки, даже когда заказ найден.

flatMap параллелит, concatMap — нет

map превращает элемент в элемент синхронно; для обращения к базе или сервису он не подходит, потому что результат там — сам Mono. Для этого flatMap: на каждый элемент он подписывается на внутренний Mono, причём сразу на многие — по умолчанию до 256 одновременно — и выпускает результаты в порядке прихода. Десять цен для десяти заказов придут перемешанными, и склеить их с заказами по позиции нельзя. concatMap подписывается по одному и порядок сохраняет, ценой последовательного выполнения; flatMapSequential даёт и параллельность, и порядок — держит буфер и выпускает элементы по номеру. Mono.zip(a, b, c) ждёт все три и отдаёт кортеж — так делают независимые вызовы параллельно в одном запросе.

Ошибка в Reactor — тоже сигнал. Исключение, брошенное внутри лямбды map, никуда не «вылетает»: оператор ловит его и отправляет вниз как onError, после чего последовательность мертва — новых элементов не будет. onErrorReturn и onErrorResume подменяют этот сигнал значением или другой цепочкой, retry подписывается заново. Порядок операторов значим: timeout(2s) перед retryWhen даёт по две секунды на каждую попытку, после него — две секунды на все попытки вместе.

Обратное давление: получатель сам говорит, сколько ему отдать

Запрос на выгрузку отдаёт миллион строк из базы клиенту, который читает их по сети в десять раз медленнее, чем база выдаёт. В блокирующем мире это решается само: поток встал на записи в сокет, следующую строку у курсора не попросил, база подождала. В мире, где никто не блокируется, ждать некому — строки копились бы в памяти, пока не кончится куча. Договор такой: подписчик вызывает request(n), и источник имеет право выдать не больше n элементов, пока не попросят ещё. Это и называется обратным давлением (backpressure).

В WebFlux цепочка договорённостей выстроена до самой базы: Reactor Netty запрашивает у Flux столько элементов, сколько влезает в буфер сокета; R2DBC-драйвер в ответ читает у курсора порцию, а остаток ждёт на стороне PostgreSQL. Память сервиса не зависит от размера выгрузки. У flatMap те же правила внутри: он берёт у источника по 32 элемента наперёд и держит не больше 256 внутренних подписок; оба числа задаются вторым и третьим аргументом.

Ломается договор на источниках, которые ждать не умеют: Flux.interval тикает по часам, событие из очереди приходит, когда пришло. Если подписчик не успел попросить, interval падает с OverflowException: Could not emit tick 0 due to lack of requests. Для таких источников стратегию выбирают явно: onBackpressureBuffer(1000) копит до предела и потом падает, onBackpressureDrop() выбрасывает лишнее, onBackpressureLatest() оставляет последнее. Это решение о том, что дороже — память, полнота или свежесть.

В MVC с виртуальными потоками обратное давление бесплатно — поток блокируется на записи в сокет; не бесплатна память: List<Order> на миллион строк соберётся целиком, если репозиторий не возвращает Stream с заданным размером выборки.

subscribeOn и publishOn: на каком потоке выполняется цепочка

По умолчанию оператор выполняется на том потоке, на котором пришёл сигнал. Для запроса в WebFlux это поток цикла событий Netty; ответ базы через R2DBC приходит с потока драйвера, ответ WebClient — с потока клиента. Так цепочка кочует между потоками, и ни один из них не ваш; управляют этим два оператора.

publishOn(scheduler) переключает поток для всего, что стоит ниже него в цепочке; место в цепочке решает всё. subscribeOn(scheduler) переключает поток, на котором произойдёт подписка, — то есть на котором начнёт работать источник, самый верх цепочки. Где он стоит, значения не имеет, а если их несколько, действует ближайший к источнику. Поэтому subscribeOn — инструмент для источника («запусти этот блокирующий вызов вон там»), publishOn — для потребителя («дальше считай на другом пуле»).

Пулов в Reactor два. Schedulers.parallel() — по потоку на ядро, для вычислений; его потоки помечены как неблокирующие, и block() на них запрещён. Schedulers.boundedElastic() — до десяти потоков на ядро, у каждого очередь до ста тысяч задач, простаивающий поток умирает через минуту; это единственное место, где блокирующий вызов допустим:

// так не надо: запрос выполнится при сборке, на потоке цикла событий
public Mono<Report> report(UUID id) {
    return Mono.just(jdbc.queryForObject(SQL, Report.class, id));
}

// так надо: вызов отложен до подписки и увезён на пул для блокирующих задач
public Mono<Report> report(UUID id) {
    return Mono.fromCallable(() -> jdbc.queryForObject(SQL, Report.class, id))
        .subscribeOn(Schedulers.boundedElastic());
}

Второй вариант работает, но это пул потоков, на котором запросы ждут базу, — модель MVC, собранная руками внутри WebFlux; если таких вызовов большинство, реактивный стек не даёт ничего.

Блокирующий вызов останавливает не запрос, а четверть сервера

С самим .block() Reactor подстраховал: потоки цикла событий помечены интерфейсом NonBlocking, и вызов block() на таком потоке падает сразу — IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-2. Запрос получит 500, сервис останется жив. Разрешён block() там, где поток не помечен: на boundedElastic, в тестах, в @Scheduled-задаче.

Опасны вызовы, которых Reactor не видит: JdbcTemplate, RestTemplate, Thread.sleep, future.get(), чтение файла, ожидание на synchronized. Они молча держат поток цикла событий, а Netty закрепляет каждое соединение за одним потоком на всё время жизни соединения. Один поток, ушедший на секунду в JDBC, — это не «минус четверть пропускной способности», а четверть всех открытых соединений, замороженных на секунду, включая долгоживущее соединение балансировщика с проверкой /health. Под нагрузкой блокирующий вызов рано или поздно оказывается на всех четырёх потоках — и встаёт всё, в том числе методы, которые в базу не ходят.

Такие вызовы ловит BlockHound — библиотека io.projectreactor.tools:blockhound. Она инструментирует JVM и бросает BlockingOperationError в момент блокирующего вызова на неблокирующем потоке; на boundedElastic блокировка остаётся разрешённой. Подключают её в тестах — один вызов BlockHound.install() в @BeforeAll базового класса — и любой интеграционный тест контроллера, где кто-то дотянулся до JDBC, начинает падать с точным стек-трейсом. На JDK 13 и новее нужен флаг JVM -XX:+AllowRedefinitionToAddDeleteMethods, иначе агент не установится. В прод BlockHound не берут: инструментирование стоит производительности.

Где ловить ошибки

Внутри цепочки ошибку обрабатывают операторы: onErrorResume подменяет упавший вызов запасным, onErrorMap переводит исключение драйвера в доменное. На границе контроллера работает то же, что в MVC: @RestControllerAdvice с @ExceptionHandler ловит и исключение, брошенное из метода напрямую, и сигнал onError из возвращённого Mono — для обработчика разницы нет. Реактивный ResponseEntityExceptionHandler отдаёт ProblemDetail, если включить spring.webflux.problemdetails.enabled=true.

Отличие от MVC — в том, что до контроллера не дошло: ошибка в WebFilter, запрос без подходящего маршрута, исключение из функционального эндпоинта. Сервлетного /error, куда MVC пересылает такие случаи, здесь нет; их обрабатывает цепочка WebExceptionHandler, и Spring Boot ставит туда свой DefaultErrorWebExceptionHandler с порядком -1, который собирает такой же JSON, что и в MVC. Свой формат — это бин, унаследованный от AbstractErrorWebExceptionHandler, с @Order(-2), чтобы встать раньше.

И одна ловушка, которой в MVC не бывает, — потоковый ответ. Когда Flux уже начал отдавать события, заголовки с кодом 200 ушли клиенту; ошибка в середине потока не превратится ни в 500, ни в ProblemDetail — соединение просто закроется, и клиент увидит оборванный поток. Обработку ошибок для таких эндпоинтов кладут внутрь Flux, до отправки.

Глубже: ThreadLocal, MDC и трассировка теряются между потокамирасширенное

Раз цепочка кочует между потоками, всё, что жило в ThreadLocal, по дороге пропадает: traceId из MDC перестаёт попадать в логи со второго оператора, SecurityContextHolder пуст, транзакция не видна. В WebFlux у этих вещей другой дом — Reactor Context: неизменяемая карта, привязанная к подписке, а не к потоку. Заполняют её снизу цепочки, contextWrite, а читают выше, deferContextual — направление обратное привычному, потому что подписка идёт снизу вверх:

public Mono<Order> get(UUID id) {
    return Mono.deferContextual(ctx -> repo.findById(id)
            .doOnNext(o -> log.info("traceId={} order={}", ctx.get("traceId"), o.id())))
        .contextWrite(ctx -> ctx.put("traceId", MDC.get("traceId")));
}

Переписывать так каждый лог не нужно. Библиотека io.micrometer:context-propagation переносит ThreadLocal в Reactor Context и обратно автоматически, если объявить, что переносить. Версии у неё свои, 1.x: её выпустили вместе с Micrometer 1.10, и искать context-propagation:1.10 бесполезно — такой версии нет. Регистрация идёт в глобальный реестр, а не в бин: ContextRegistry.getInstance() возвращает один объект на всю JVM, и в него дописывают свой accessor при старте.

import io.micrometer.context.ContextRegistry;
import jakarta.annotation.PostConstruct;
import org.slf4j.MDC;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ContextPropagationConfig {

    @PostConstruct
    void registerTraceId() {
        ContextRegistry.getInstance()
            .registerThreadLocalAccessor("traceId",
                () -> MDC.get("traceId"),
                v -> MDC.put("traceId", v),
                () -> MDC.remove("traceId"));
    }
}

Со Spring Boot 3.2 и этого чаще всего не нужно: если библиотека лежит в проекте, достаточно строки spring.reactor.context-propagation=auto. По умолчанию стоит limited, и тогда ThreadLocal восстанавливается лишь внутри двух операторов, tap и handle. Трассировка через Micrometer Tracing едет на том же механизме, и идентификатор трассы в логах подчиняется той же настройке.

Глубже: R2DBC: база без блокировок и без JPAрасширенное

JDBC блокирующий по замыслу: ResultSet.next() ждёт данных на потоке. Для WebFlux нужен драйвер, который отдаёт строки сигналами, — это R2DBC; есть для PostgreSQL, MySQL, MSSQL, H2. Поверх него Spring Data даёт привычные репозитории:

public interface OrderRepository extends R2dbcRepository<Order, UUID> {

    @Query("SELECT * FROM orders WHERE customer_id = :customerId")
    Flux<Order> findByCustomer(UUID customerId);

    Mono<Order> findByOrderNumber(String orderNumber);
}

Привычность заканчивается на сигнатурах. R2DBC — не Hibernate: R2dbcEntityTemplate ближе к JdbcTemplate, чем к EntityManager. Связей между сущностями нет — @OneToMany не существует, вложенные объекты собирают вторым запросом руками; ленивой загрузки, отслеживания изменений и каскадов тоже нет: что изменили, то и сохраняете явно. Транзакции работают: @Transactional на методе, возвращающем Mono, идёт через R2dbcTransactionManager, а сама транзакция живёт в Reactor Context; для программного управления есть TransactionalOperator. Но на методе с обычным возвращаемым типом реактивный менеджер откажет: Cannot apply reactive transaction to non-reactive return type.

Отсюда практика: на записи, где создают и меняют заказ, проверяют правила и отслеживают изменения сущностей, R2DBC не берут — без возможностей JPA этот код становится громоздким. На чтении берут, когда данные надо отдавать потоком по мере готовности.

Глубже: WebClient, RestClient и что стало с RestTemplateрасширенное

Частое заблуждение: RestTemplate объявлен устаревшим. Аннотации @Deprecated на нём нет — его просто перестали развивать: новых возможностей там не будет, переписывать рабочий код ради этого не нужно.

WebClient — неблокирующий клиент, родной для WebFlux; работает и в блокирующем стеке. Таймаут, повторы и перевод статусов в исключения объявляются в той же цепочке, что и запрос:

@Configuration
public class PricingClientConfig {

    @Bean
    public WebClient pricingClient(WebClient.Builder builder) {
        return builder
            .baseUrl("https://pricing.internal")
            .defaultHeader("X-Service", "orders")
            .build();
    }
}

@Component
@RequiredArgsConstructor
public class PricingClient {
    private final WebClient client;

    public Mono<Price> calculate(Order order) {
        return client.post().uri("/quote")
            .bodyValue(order)
            .retrieve()
            .onStatus(HttpStatusCode::is4xxClientError, resp -> Mono.error(new BadRequest()))
            .onStatus(HttpStatusCode::is5xxServerError, resp -> Mono.error(new PricingDownException()))
            .bodyToMono(Price.class)
            .timeout(Duration.ofSeconds(2))
            .retryWhen(Retry.backoff(3, Duration.ofMillis(200)));
    }
}

Порядок timeout → retryWhen даёт каждой из трёх попыток свои две секунды. А совет «в MVC берите WebClient и делайте .block()» устарел: на потоках Tomcat block() разрешён, но с Spring Framework 6.1 (Spring Boot 3.2) есть RestClient — синхронный клиент с тем же текучим API, только без Mono. Для блокирующего стека берут его: пишется как WebClient, читается как обычный код.

Price price = restClient.post().uri("/quote")
    .body(order)
    .retrieve()
    .body(Price.class);

Глубже: стек-трейс без вашего кода: checkpoint и debugрасширенное

Исключение из реактивной цепочки приходит со стек-трейсом, в котором нет ни одной вашей строки — только MonoFlatMap, FluxMap, FluxOnErrorResume. Причина — разделение сборки и выполнения: ваш код отработал при сборке, упало при выполнении на другом потоке, и стек этого потока о вашей строке не знает.

Три средства, по нарастающей. .checkpoint("after-pricing") в цепочке добавляет к ошибке метку — дёшево, но метки расставляют руками. Hooks.onOperatorDebug() при старте запоминает стек сборки для каждого оператора, и ошибка приходит с полной картой цепочки; это дорого, только для локальной отладки. Для прода есть reactor-tools с ReactorDebugAgent: агент инструментирует байт-код операторов при старте и записывает место сборки без затрат во время работы. Spring Boot включает его сам, если зависимость лежит в проекте; выключается через spring.reactor.debug-agent.enabled=false.

Глубже: тесты: StepVerifier и WebTestClientрасширенное

assertEquals с Mono сравнил бы описание, а не результат; block() в тесте допустим, но видит только конечное значение. StepVerifier из reactor-test подписывается сам и проверяет сигналы по порядку:

StepVerifier.create(service.prices(List.of(1L, 2L)))
    .expectNext(new Price(1L, 100), new Price(2L, 250))
    .verifyComplete();

Без verify… в конце ничего не произойдёт — это и есть подписка. Для цепочек с задержками и повторами есть виртуальное время: StepVerifier.withVirtualTime(() -> client.calculate(order)) и .thenAwait(Duration.ofSeconds(10)) прокручивают часы Reactor, и тест retryWhen с растущей паузой проходит за миллисекунды. Контроллеры проверяют через @WebFluxTest и WebTestClient — аналог MockMvc с текучим API: client.get().uri("/api/v1/orders/{id}", id).exchange().expectStatus().isOk(). Туда же ставят BlockHound.

Коротко

• По умолчанию — MVC с виртуальными потоками. WebFlux — для потоковой отдачи, тысяч открытых соединений и обратного давления, и только при неблокирующем стеке целиком; два стартера в одном модуле — это MVC.
• Mono и Flux — описание, а не значение: до subscribe() ничего не выполняется, подписывается WebFlux. Mono.just вычисляет аргумент сразу, fromCallable и defer — при подписке; каждая подписка выполняет цепочку заново.
• flatMap параллелит до 256 подписок и перемешивает порядок, concatMap и flatMapSequential его хранят. Ошибка — сигнал, после которого поток мёртв; retry повторяет с начала.
• Обратное давление — request(n) от подписчика до курсора базы; на источниках, которые не умеют ждать, стратегию onBackpressure* выбирают явно.
• publishOn — для того, что ниже; subscribeOn — для источника. boundedElastic — единственное место для блокирующего кода, и это MVC, собранный руками.
• block() на цикле событий падает сразу; JDBC и другие невидимые Reactor вызовы замораживают четверть соединений вместе с /health. Ловит BlockHound в тестах.
• Вместо ThreadLocal — Reactor Context; context-propagation и spring.reactor.context-propagation=auto возвращают MDC и трассировку в логи.
• @RestControllerAdvice работает; что не дошло до контроллера — в ErrorWebExceptionHandler; начатый потоковый ответ 500 уже не вернёт.
• R2DBC — без связей, ленивой загрузки и отслеживания изменений; на запись обычно остаются JPA и MVC.
• Тесты — StepVerifier с виртуальным временем и WebTestClient; стек-трейс без вашего кода чинят checkpoint и ReactorDebugAgent.

Что почитать дальше

• Spring MVC — блокирующий стек, которого чаще всего достаточно, и обработка ошибок для сравнения.
• Spring MVC, WebFlux и виртуальные потоки — тот же выбор со стороны архитектуры сервиса, не кода.
• Виртуальные потоки Java 21 — как устроены парковка и прилипание.
• @Transactional глубоко — транзакция в ThreadLocal в MVC и почему в WebFlux она переехала в контекст.