Reactive Programming with Project Reactor

This section documents Spring Boot 4.1.x and Spring Framework 7.0.x on the Java 17/21+ baseline, as described by the official Spring Boot reference documentation and each sub-project’s own site (Spring Data, Spring for Apache Kafka, Micrometer, Project Reactor, springdoc-openapi, Spring gRPC, and the others listed in this section’s Bibliography) — which are the references these pages are written and verified against.

This content was generated with the assistance of AI and should be verified against those official docs before being relied on in production. Spring Boot and its ecosystem continue to evolve; the examples here target the current 4.1.x / 7.0.x releases.

This section’s bibliography lists the reference material consulted while preparing these pages.

Spring WebFlux, R2DBC, and reactive Spring Data are all built on top of Project Reactor, the Reactive Streams implementation used throughout the Spring ecosystem. This page introduces Reactor’s two core types, its two-phase execution model, the operators used to compose asynchronous pipelines, and how to test and troubleshoot reactive code.

Mono and Flux

Reactor exposes two publishers. Mono<T> emits at most one element (or an error, or completes empty) — the reactive analogue of Optional<T> or a CompletableFuture<T>. Flux<T> emits zero to many elements over time — the reactive analogue of a Stream<T> or a List<T>, except items may arrive asynchronously.

Mono<String> mono = Mono.just("hello");
Mono<String> empty = Mono.empty();
Mono<String> failed = Mono.error(new IllegalStateException("boom"));

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> ranged = Flux.range(1, 5);
Flux<String> fromIterable = Flux.fromIterable(List.of("a", "b", "c"));

// bridging existing async APIs
Mono<String> fromFuture = Mono.fromFuture(() -> someCompletableFuture());
Mono<String> fromCallable = Mono.fromCallable(() -> blockingCall());

Both types implement org.reactivestreams.Publisher, so they interoperate with any other Reactive Streams library. See the Project Reactor reference documentation for the full API surface and the reactive programming background it builds on.

Assembly vs. subscription

A Reactor pipeline is built in two distinct phases, and understanding the split explains most of the surprises newcomers hit.

Assembly happens when a chain of operators is declared: Flux.just(…​).map(…​).filter(…​) returns a new, immutable publisher describing the pipeline, but nothing has executed yet. Operators return new instances rather than mutating the receiver — forgetting to reassign or chain the result is the most common Reactor bug.

Subscription happens only when a Subscriber calls subscribe() (directly, or indirectly through a framework such as WebFlux handling an HTTP request, or a test driver such as StepVerifier). Only at that point does data actually start flowing, pulled by the subscriber according to its declared demand.

// ASSEMBLY: nothing runs yet, this only builds a description of the pipeline
Flux<String> pipeline = Flux.just("a", "b", "c")
        .map(String::toUpperCase)
        .doOnNext(s -> System.out.println("assembled but not yet subscribed: " + s));

// common mistake: this line does nothing useful, the "filtered" result is discarded
pipeline.filter(s -> s.startsWith("A"));

// SUBSCRIPTION: only now does anything actually execute
pipeline.subscribe(System.out::println);

// a cold publisher replays its whole sequence for every new subscriber
pipeline.subscribe(s -> System.out.println("second subscriber: " + s));

Mono/Flux are cold publishers by default: each subscription re-runs the assembled pipeline from scratch. Operators such as share() and cache() convert a cold sequence into a hot one shared across subscribers.

Core operators

map and flatMap

map applies a synchronous, one-to-one transformation. flatMap applies a function that itself returns a publisher, and flattens (merges) the resulting inner publishers into the output sequence — it is the operator to reach for whenever the transformation is itself asynchronous (a database call, an HTTP request).

Flux<Integer> lengths = Flux.just("reactor", "spring", "boot")
        .map(String::length);                       // synchronous, 1-to-1

Flux<Order> ordersWithItems = Flux.just("order-1", "order-2")
        .flatMap(orderId -> orderRepository.findById(orderId)  // returns Mono<Order>, async
                .flatMap(order -> itemRepository.findByOrderId(order.getId())
                        .collectList()
                        .map(items -> {
                            order.setItems(items);
                            return order;
                        })));

flatMap does not preserve ordering between inner publishers (they are merged as they complete); concatMap behaves like flatMap but preserves order at the cost of concurrency.

zip

zip combines the corresponding elements of two or more publishers pairwise, completing (or erroring) as soon as any source does:

Mono<User> user = userRepository.findById(userId);
Mono<List<Order>> orders = orderRepository.findByUserId(userId).collectList();

Mono<UserProfile> profile = Mono.zip(user, orders)
        .map(tuple -> new UserProfile(tuple.getT1(), tuple.getT2()));

merge and concat

merge subscribes to all sources eagerly and interleaves their emissions as they arrive (no ordering guarantee); concat subscribes to each source in turn, only after the previous one completes, preserving source order:

Flux<String> merged = Flux.merge(
        Flux.interval(Duration.ofMillis(100)).map(i -> "fast-" + i).take(3),
        Flux.interval(Duration.ofMillis(150)).map(i -> "slow-" + i).take(3));

Flux<String> concatenated = Flux.concat(
        Flux.just("first-a", "first-b"),
        Flux.just("second-a", "second-b"));   // only starts after the first Flux completes

Backpressure

Reactive Streams is a pull-based protocol: a subscriber signals how many elements it can currently handle via request(n), and a well-behaved publisher never emits more than that outstanding demand. This is what lets a slow consumer avoid being overwhelmed by a fast producer without buffering unboundedly.

Flux.range(1, 1_000_000)
        .doOnRequest(n -> System.out.println("requested: " + n))
        .subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(10);                       // initial demand
            }

            @Override
            protected void hookOnNext(Integer value) {
                process(value);
                if (value % 10 == 0) {
                    request(10);                   // request the next batch once this one is processed
                }
            }
        });

// declarative alternative: cap how much is buffered/prefetched by an operator
Flux.range(1, 1_000_000)
        .onBackpressureBuffer(1_000)               // bound the buffer, drop or error past the limit
        .publishOn(Schedulers.boundedElastic(), 256);

Most application code never implements Subscription directly — operators such as onBackpressureBuffer, onBackpressureDrop, and onBackpressureLatest, or simply sizing `publishOn’s prefetch, are enough to keep demand under control.

Schedulers and context

By default Reactor executes on whichever thread called subscribe() (often the framework’s I/O thread, e.g. a Netty event-loop thread under WebFlux). publishOn switches the thread used by downstream operators from that point on; subscribeOn affects where the subscription itself, and therefore the upstream source, runs.

Mono<String> result = Mono.fromCallable(() -> blockingLegacyCall())   // a blocking call
        .subscribeOn(Schedulers.boundedElastic())   // run the blocking source off the event loop
        .map(this::transform)
        .publishOn(Schedulers.parallel())           // switch downstream processing to a CPU-bound pool
        .doOnNext(this::record);

Reactor ships four built-in Scheduler factories: Schedulers.immediate() (no switch), Schedulers.single() (one reusable thread), Schedulers.parallel() (a fixed pool sized to available CPUs, for CPU-bound work), and Schedulers.boundedElastic() (a bounded, growable pool for blocking or I/O-bound calls that cannot be avoided). Never block inside a parallel() or event-loop thread — wrap unavoidable blocking calls in boundedElastic().

Because a reactive pipeline may hop across several threads, Reactor cannot rely on ThreadLocal to carry request-scoped state (a trace ID, a security principal). Instead it propagates an immutable Context map alongside the signals, written top-down from the nearest downstream contextWrite and readable anywhere upstream:

Mono<String> withContext = Mono.deferContextual(contextView ->
        Mono.just("hello, " + contextView.get("user")))
        .contextWrite(Context.of("user", "alice"));    // visible to operators upstream of this point

Reactive transactions and thread affinity

Imperative Spring transactions are thread-bound. JpaTransactionManager and DataSourceTransactionManager bind the JDBC Connection to the calling thread through TransactionSynchronizationManager, which holds it in a ThreadLocal; every @Transactional method further down the call stack finds the same connection because it runs on the same thread. A reactive pipeline offers no such guarantee. Reactive transaction managers — ReactiveTransactionManager implementations such as R2dbcTransactionManager or ReactiveMongoTransactionManager — therefore bind the transactional resources to the Context described above instead of to a thread. As the Spring team puts it, "Reactor Context is to reactive programming what ThreadLocal is to imperative programming": the Context carries the transaction state, its resources and its synchronizations, and because a Context belongs to a single Subscription, the transaction covers exactly one subscription of one chain.

This section is about whether a given statement takes part in the transaction at all. For the orthogonal questions of what a transaction is allowed to see and how competing writers are serialised, see Transaction Isolation & Locking.

Anything that leaves the chain leaves the transaction. The common ways to do that by accident:

  • An inner .subscribe(). Calling subscribe() on a nested publisher starts a new subscription with a fresh, empty Context, so that work never sees the transaction.

  • A publisher fired from doOnNext or doOnEach. These callbacks exist for side effects on the signal, not for composing work; a publisher created inside one is either never subscribed to at all or subscribed to separately, and neither outcome is part of the unit of work.

  • A CompletableFuture bridged in by hand. A future that is already running when the chain is assembled executes on its own executor. Wrap it as Mono.fromFuture(() → …​) so the call is deferred to subscription time and stays inside the chain.

  • @Async methods. The proxy hands the call to a TaskExecutor thread that has neither the caller’s ThreadLocal state nor its Reactor Context.

  • A blocking @Transactional proxy method that returns a publisher. With a PlatformTransactionManager the interceptor commits when the method returns — and what it returns is an unsubscribed publisher, so the transaction opens and commits around a value that has not touched the database yet. The data access then runs afterwards, outside any transaction.

In each case a later failure rolls back only the work that stayed inside the chain. The detached write has committed independently and survives the rollback.

The failure mode is breaking the chain, not switching threads. publishOn, subscribeOn and parallel() all change which thread executes an operator, but none of them starts a new subscription, so the Context — and with it the transaction — crosses the hop untouched. A single correctly composed chain stays transactional however many schedulers it visits.

Concurrency within the chain is a different matter. flatMap subscribes to up to 256 inner publishers at once by default, while an R2DBC transaction runs over a single connection, so those inner statements end up serialised onto that one connection in a non-deterministic order. Inside a transaction prefer concatMap, or flatMap(fn, 1), so the statements execute in the order they were written.

// TransactionalOperator.create(reactiveTransactionManager)
private final TransactionalOperator txOperator;

Mono<Void> transferFunds(String from, String to, BigDecimal amount) {
    Mono<Void> unitOfWork = accounts.debit(from, amount)
            .then(accounts.credit(to, amount))          // both writes, one chain, one Context
            .then();

    return unitOfWork.as(txOperator::transactional)     // commits on completion; rolls back on error
            .onErrorResume(InsufficientFundsException.class,   // handled AFTER the operator, so the
                    ex -> auditLog.record(from, ex));          // rollback has already happened
}

The same unit of work can be expressed declaratively: @Transactional on a method whose return type is a Publisher routes to Spring’s reactive transaction management, which needs a ReactiveTransactionManager bean rather than a PlatformTransactionManager. Since Spring Framework 5.3 a cancel signal also triggers a rollback, so a Flux inside a transaction must be consumed in full for that transaction to commit — truncating it with take(n) downstream of the transactional operator rolls the whole thing back.

Written the wrong way, the second write escapes:

Mono<Void> brokenTransfer(String from, String to, BigDecimal amount) {
    return accounts.debit(from, amount)
            .doOnNext(debited ->
                    accounts.credit(to, amount).subscribe())    // WRONG: a new subscription with an
            .then()                                             // empty Context -- no transaction
            .as(txOperator::transactional);
}

The credit runs on its own subscription, outside the transaction. If the debit is rolled back the credit is not, and the two accounts silently disagree.

Bridging to blocking code safely

The rule stated above — never block a parallel() or event-loop thread — is one Reactor partly enforces on its own. Calling block(), blockFirst() or blockLast() from a thread created by Schedulers.parallel() or Schedulers.single() fails immediately:

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in
thread reactor-http-nio-3

Those threads implement Reactor’s NonBlocking marker interface, and so do Netty’s event-loop threads under WebFlux — which is why this exception typically surfaces with a reactor-http-* thread name in a request handler rather than in a scheduler of your own making.

Where blocking is not forbidden it is still hazardous. boundedElastic() is a capped pool — by default ten threads per CPU core, with a queue of up to 100 000 further tasks behind it — so a task that blocks while waiting on work that itself needs a thread from that same pool holds its own thread hostage. Under load enough of these starve the pool and deadlock it. On an event loop the failure is faster and total: a few threads serve all I/O for the entire application, so a blocked event-loop thread cannot process the very response it is waiting for, and the wait can never end.

flowchart TD Need["Need a value from\na Mono or Flux"] --> Compose{"Can you compose it?\nflatMap / zip / expand"} Compose -->|Yes| Chain["Stay in the chain.\nNever call block()"] Compose -->|"No: legacy blocking library"| Offload["Mono.fromCallable(...)\n.subscribeOn(boundedElastic())"] Compose -->|"No: non-reactive SPI\ndemands a value now"| Guarded["Dedicated Scheduler\n+ subscribeOn\n+ toFuture().get(timeout)"] Guarded --> Which{"Which thread is\nthe caller on?"} Which -->|"Event loop or parallel()"| Refuse["Unsafe. Move the boundary\noutwards instead"] Which -->|Worker thread| Last["Acceptable last resort"]

In order of preference:

  1. Compose instead of bridging. flatMap, zip, then and expand express nearly every "I need this value before I can continue" case without ever leaving the reactive world.

  2. Offload an unavoidable blocking client. Wrap it as Mono.fromCallable) → legacyBlockingCall(.subscribeOn(Schedulers.boundedElastic()), which keeps the blocking call on a pool built for exactly that and off every non-blocking thread.

  3. In tests, use StepVerifier rather than block() — see Testing reactive streams with StepVerifier below. Tests are where block() creeps in most easily, and a StepVerifier asserts more precisely anyway.

  4. Install BlockHound to catch the accidents. BlockHound (io.projectreactor.tools:blockhound, a separate artifact from reactor-tools) instruments the JVM so that any blocking call made from a thread marked NonBlocking throws a BlockingOperationError pointing at the offending line. A single BlockHound.install() in a development or test profile turns a class of production-only latency bugs into loud, local failures.

When a non-reactive SPI genuinely demands a value synchronously and none of the above applies, the bridge must be guarded rather than improvised:

// A dedicated pool: exhausting it degrades this adapter only, never the shared boundedElastic().
private final Scheduler bridgeScheduler =
        Schedulers.newBoundedElastic(16, 64, "legacy-spi-bridge");

@PreDestroy
void shutdown() {
    bridgeScheduler.dispose();
}

// Precondition: never called from an event-loop or parallel() thread.
Report renderReport(String id) throws Exception {
    return reportService.build(id)          // the inner flow subscribes on our own pool,
            .subscribeOn(bridgeScheduler)   // so it cannot be blocked by the caller's thread
            .toFuture()
            .get(5, TimeUnit.SECONDS);      // always bounded -- never get() with no timeout
}

Four precautions make this pattern survivable, and all four are load-bearing. The pool is dedicated, so saturating it cannot take down unrelated blocking work sharing boundedElastic(). The method is never invoked from a non-blocking thread, which is the precondition Reactor cannot check for you here because get() is Future’s method, not Reactor’s, and therefore raises no `IllegalStateException. Pool exhaustion is treated as backpressure — reject the request with a 503 rather than queueing without limit. And the reactive Context does not cross the toFuture() boundary, so anything riding in it, transaction context very much included, is gone: the inner flow needs its own contextWrite, and it can never join a transaction owned by the caller.

Error handling

Errors are a terminal signal in Reactive Streams: once onError fires, the sequence is over unless an operator intercepts and replaces it.

Mono<Order> order = orderRepository.findById(orderId)
        .switchIfEmpty(Mono.error(new OrderNotFoundException(orderId)))
        .onErrorResume(OrderNotFoundException.class,
                ex -> Mono.just(Order.placeholderFor(orderId)))    // fall back to a default value
        .onErrorResume(TimeoutException.class,
                ex -> Mono.error(new ServiceUnavailableException(ex)))  // translate to a different error
        .doOnError(ex -> log.error("failed to load order {}", orderId, ex));

Flux<String> resilient = externalServiceClient.streamUpdates()
        .retryWhen(Retry.backoff(3, Duration.ofMillis(200))
                .filter(ex -> ex instanceof TransientException))    // only retry recoverable failures
        .onErrorReturn("unavailable");                              // last-resort fallback value

onErrorReturn supplies a static fallback value, onErrorResume supplies a fallback publisher (so it can itself be asynchronous), and onErrorMap translates one exception type into another without swallowing the error. retry(n) re-subscribes immediately up to n times; retryWhen(Retry.backoff(…​)) (from reactor-extra’s `reactor.util.retry.Retry) adds exponential backoff and jitter, and should be scoped with .filter(…​) so only genuinely transient failures are retried.

Testing reactive streams with StepVerifier

StepVerifier subscribes to a Publisher and asserts, step by step, exactly which signals it emits — values, completion, or an error — making reactive pipelines as testable as synchronous code.

@Test
void mapsAndCompletes() {
    Flux<String> flux = Flux.just("a", "b", "c").map(String::toUpperCase);

    StepVerifier.create(flux)
            .expectNext("A", "B", "C")
            .verifyComplete();
}

@Test
void propagatesAnError() {
    Mono<Order> mono = orderRepository.findById("missing")
            .switchIfEmpty(Mono.error(new OrderNotFoundException("missing")));

    StepVerifier.create(mono)
            .expectError(OrderNotFoundException.class)
            .verify();
}

@Test
void respectsVirtualTime() {
    StepVerifier.withVirtualTime(() -> Flux.interval(Duration.ofHours(1)).take(3))
            .thenAwait(Duration.ofHours(3))   // fast-forwards virtual time instead of really waiting 3 hours
            .expectNextCount(3)
            .verifyComplete();
}

StepVerifier.withVirtualTime lets tests exercise time-based operators (delayElements, interval, timeouts) instantly instead of pausing the test thread. For asserting demand and backpressure behavior directly, PublisherProbe and a manually-driven TestPublisher complement StepVerifier in the reactor-test module.

How WebFlux, R2DBC, and reactive Spring Data build on Reactor

Reactor is the foundation the rest of Spring’s reactive stack is written against:

  • Spring WebFlux controllers and RouterFunction handlers may return Mono<T>/Flux<T> directly; the framework subscribes to them as part of handling each request, on Netty’s (or another server’s) event-loop threads.

  • R2DBC (Reactive Relational Database Connectivity) exposes SQL access through Mono/Flux-returning DatabaseClient calls instead of blocking JDBC.

  • Reactive Spring Data repositories (ReactiveCrudRepository, and the reactive Couchbase, MongoDB, and Neo4j variants covered elsewhere in this section) return Mono/Flux from their query methods, so the same operators and testing techniques on this page apply directly to repository calls.

@RestController
class OrderController {

    private final ReactiveOrderRepository repository;

    OrderController(ReactiveOrderRepository repository) {
        this.repository = repository;
    }

    @GetMapping("/orders/{id}")
    Mono<Order> findById(@PathVariable String id) {
        return repository.findById(id)
                .switchIfEmpty(Mono.error(new OrderNotFoundException(id)));
    }

    @GetMapping("/orders")
    Flux<Order> streamAll() {
        return repository.findAll();
    }
}

Because a reactive controller method, an R2DBC query, and a reactive repository call all speak the same Mono/Flux vocabulary, the assembly/subscription model, the operators, and the StepVerifier-based testing approach described above carry over unchanged from a plain Reactor pipeline to a full WebFlux request handling chain. Combining those pieces in a real request path does add two caveats that a standalone pipeline never runs into: the transaction now rides the subscription rather than the thread (Reactive transactions and thread affinity), and the handler runs on an event-loop thread where blocking is fatal (Bridging to blocking code safely).

Further learning

This page covers the core mental model, but Reactor’s operator set is large and the timing subtleties around schedulers, context, and backpressure reward hands-on practice. For guided exercises, see the Reactor workshop, and for the authoritative, continuously updated operator reference, see the Project Reactor reference documentation. If Project Reactor isn’t the right fit for a given service, see Concurrency Alternatives to Reactive Programming for a side-by-side comparison with Java Virtual Threads and Kotlin Coroutines, including how transaction management differs under each.