Reactor Core: Mono and Flux

Introduction

Reactive programming models async data streams with backpressure—consumers signal how much data they can handle. Project Reactor provides Mono (0–1 item) and Flux (0–N items), the foundation of Spring WebFlux and R2DBC. This chapter learns operators, threading, and debugging before building reactive web apps.

Prerequisites

Spring MVC Basics (compare with blocking style)
Java 17+ and comfort with lambdas

Blocking vs Reactive Mental Model

Blocking:

java

String result = restClient.get().uri("/api").retrieve().body(String.class);

Thread blocked until response arrives.

Reactive:

java

Mono<String> result = webClient.get().uri("/api").retrieve().bodyToMono(String.class);

Thread free until subscription; result flows when data ready.

Mono and Flux Basics

java

import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
 
Mono<String> one = Mono.just("Hello");
Flux<String> many = Flux.just("a", "b", "c");
Flux<Integer> range = Flux.range(1, 5);

Subscribe and print (tests):

java

one.subscribe(System.out::println);
 
many.map(String::toUpperCase)
    .filter(s -> s.startsWith("A"))
    .subscribe(System.out::println);

Code explanation:

map transforms elements
filter skips non-matching
Nothing runs until subscribe (lazy)

zip and flatMap

Combine two async sources:

java

Mono<User> user = fetchUser(1);
Mono<List<Order>> orders = fetchOrders(1);
 
Mono<UserOrders> combined = Mono.zip(user, orders)
    .map(tuple -> new UserOrders(tuple.getT1(), tuple.getT2()));

Chain async calls:

java

Mono<Profile> profile = userIdMono
    .flatMap(this::fetchProfile);

flatMap flattens nested Mono<Mono<T>> into Mono<T>—essential for sequential API calls.

Threading and Schedulers

By default many operators run on caller thread. Offload blocking work:

java

Mono.fromCallable(() -> blockingDbCall())
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe();

Scheduler	Use
parallel	CPU-bound work
boundedElastic	Blocking IO (legacy JDBC in reactive app—avoid mixing)
immediate	Current thread

Never block inside reactive chain on event loop thread— freezes Netty.

Backpressure

Flux supports limitRate, onBackpressureBuffer, onBackpressureDrop when producer faster than consumer.

java

Flux.range(1, 1_000_000)
    .limitRate(100)
    .subscribe(System.out::println);

Error Handling

java

Mono.error(new RuntimeException("fail"))
    .onErrorReturn("fallback")
    .subscribe(System.out::println);
 
Flux.just(1, 2, 0, 4)
    .map(i -> 10 / i)
    .onErrorContinue((ex, val) -> log.warn("skip {}", val))
    .subscribe(System.out::println);

onErrorMap wraps exceptions for API layer.

Debugging

Enable operator debug:

java

Hooks.onOperatorDebug();

Log signals:

java

flux.doOnNext(v -> log.info("next {}", v))
    .doOnError(ex -> log.error("error", ex))
    .doOnComplete(() -> log.info("complete"))
    .subscribe();

Use StepVerifier in tests:

java

StepVerifier.create(Flux.just("a", "b"))
    .expectNext("a", "b")
    .verifyComplete();

When to Adopt Reactive

Good fit	Poor fit
High concurrency IO, streaming	JDBC-heavy CRUD team unfamiliar with reactive
Gateway aggregating many services	Simple internal admin APIs
End-to-end reactive stack	"Reactive rewrite" without bottleneck proof

Most hello_code apps stay Spring MVC + blocking JDBC until requirements justify complexity.

FAQ

Mono empty?

Mono.empty() completes without value—handle with switchIfEmpty.

Block in test?

mono.block() allowed in tests—not in production reactive threads.

Hot vs cold publisher?

Cold starts on subscribe; hot (share) broadcasts—share(), cache().

Iterable to Flux?

Flux.fromIterable(list).

Kotlin coroutines?

kotlinx-coroutines-reactor bridges awaitSingle().

Next chapter?

Spring WebFlux Basics builds HTTP on Reactor.