Reactive Programming with Project Reactor

Traditional Spring MVC code is blocking: a thread handles a request and waits while the database or a remote API responds. Reactive programming takes the opposite approach: work is described as a pipeline of asynchronous steps, and threads are never parked waiting for I/O. A small number of threads can then serve a very large number of concurrent, slow connections — streaming, chat, gateways, and fan-out to many services.

Spring's reactive stack is built on Project Reactor. Before touching WebFlux you need to be comfortable with its two types, Mono and Flux, and its operators. This lesson covers the reactive-streams model, creating and transforming publishers, combining streams, error handling, backpressure, schedulers, and testing with StepVerifier.

Mono and Flux

A Mono<T> emits zero or one value and then completes (or errors) — like an asynchronous Optional. A Flux<T> emits zero to many values — like an asynchronous stream. Both are lazy: nothing happens until someone subscribes. In a WebFlux application the framework subscribes for you when it writes the HTTP response, so your code returns publishers and almost never calls subscribe() itself.

Operators

Reactor offers hundreds of operators; these are the ones you use daily:

• map — transform each value synchronously.
• flatMap — transform each value into another publisher (an async call) and merge the results; concatMap keeps order.
• filter, take, skip, distinct — select values.
• zip / Mono.zip — combine results of several independent calls.
• switchIfEmpty, defaultIfEmpty — handle missing values.
• onErrorResume, onErrorReturn, retryWhen, timeout — resilience.
• doOnNext, doOnError, log — side effects for logging and debugging.
• collectList, reduce, buffer, window — aggregate.

Backpressure

Reactive Streams lets a subscriber tell the publisher how many items it can handle (request(n)). A slow consumer therefore cannot be flooded by a fast producer. Operators such as limitRate, onBackpressureBuffer and onBackpressureDrop let you decide what happens when a source cannot slow down.

Never Block Inside a Reactive Pipeline

WebFlux runs on a few event-loop threads (one per CPU core by default). Calling a blocking API — JDBC, Thread.sleep, RestTemplate, block() — on those threads freezes every connection they serve. If you must call blocking code, wrap it with Mono.fromCallable(...) and move it to Schedulers.boundedElastic() with subscribeOn. Tools like BlockHound detect accidental blocking in tests.

Reactive or Virtual Threads?

Since Java 21, virtual threads let ordinary blocking Spring MVC code scale to many concurrent requests with far less complexity. For most CRUD services, Spring MVC with virtual threads is simpler to write, debug and test. Reactive remains the better fit for streaming data, long-lived connections (SSE, WebSockets), backpressure, and composing many asynchronous sources.

Examples

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

Mono<String> name = Mono.just("asha");
Mono<String> empty = Mono.empty();
Flux<Integer> scores = Flux.just(72, 95, 88, 40, 99);

name.map(String::toUpperCase)
    .subscribe(n -> System.out.println("name: " + n));

scores.filter(s -> s >= 80)
      .map(s -> s + " (pass with distinction)")
      .subscribe(System.out::println);

empty.defaultIfEmpty("anonymous").subscribe(n -> System.out.println("user: " + n));

Flux.range(1, 5)
    .reduce(0, Integer::sum)
    .subscribe(sum -> System.out.println("sum 1..5 = " + sum));

name: ASHA
95 (pass with distinction)
88 (pass with distinction)
99 (pass with distinction)
user: anonymous
sum 1..5 = 15

Mono<User> findUser(long id) { ... }            // async lookups
Mono<List<Order>> ordersOf(long userId) { ... }
Mono<Integer> pointsOf(long userId) { ... }

record Dashboard(User user, List<Order> orders, int points) {}

Mono<Dashboard> dashboard(long id) {
    return findUser(id)
        .switchIfEmpty(Mono.error(new UserNotFoundException(id)))
        .flatMap(user -> Mono.zip(ordersOf(user.id()), pointsOf(user.id()))   // both calls in parallel
            .map(t -> new Dashboard(user, t.getT1(), t.getT2())))
        .timeout(Duration.ofSeconds(2))
        .onErrorResume(TimeoutException.class, e -> Mono.error(new ServiceUnavailableException("Dashboard timed out")));
}

// Flux of ids -> fetch each user concurrently (max 4 at a time)
Flux<User> users = Flux.just(1L, 2L, 3L, 4L, 5L)
    .flatMap(this::findUser, 4);

dashboard(7)   -> Dashboard[user=User[id=7, name=Asha], orders=[...], points=150]
dashboard(999) -> error UserNotFoundException
(slow services) -> error ServiceUnavailableException: Dashboard timed out

// A legacy blocking library call
LegacyReport generateBlocking(long id) { ... }        // takes ~1 s, blocks the thread

Mono<LegacyReport> generate(long id) {
    return Mono.fromCallable(() -> generateBlocking(id))
               .subscribeOn(Schedulers.boundedElastic());   // runs on a thread pool meant for blocking work
}

[boundedElastic-1] generating report 42
(event-loop threads stay free to serve other requests)

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
</dependency>

@Test
void filtersHighScores() {
    Flux<Integer> high = Flux.just(72, 95, 88, 40).filter(s -> s >= 80);

    StepVerifier.create(high)
        .expectNext(95, 88)
        .verifyComplete();
}

@Test
void timesOutUsingVirtualTime() {
    StepVerifier.withVirtualTime(() -> Mono.never().timeout(Duration.ofSeconds(30)))
        .thenAwait(Duration.ofSeconds(30))
        .expectError(TimeoutException.class)
        .verify();                               // completes instantly, no real waiting
}

✔ filtersHighScores()
✔ timesOutUsingVirtualTime()   (2 ms)