Reactive programming is a paradigm for building asynchronous, non-blocking, event-driven systems. Instead of calling methods that block until a result is ready, you define a pipeline of transformations on a stream of data that flows asynchronously. Project Reactor is the reactive library at the core of Spring WebFlux. It implements the Reactive Streams specification, ensuring interoperability with other reactive libraries like RxJava and Mutiny.
Step 1 — Why Reactive? The Problem with Blocking I/O
In a traditional Spring MVC app, each HTTP request occupies one thread from a pool (typically 200 threads). When that request does database I/O or calls an external service, the thread blocks and waits. Under high load, all threads are waiting — new requests queue up. Reactive eliminates blocking: one thread handles thousands of concurrent operations by never waiting.
Traditional vs Reactive
- Traditional (blocking): 1 request = 1 thread. Thread blocks on I/O. 500 concurrent requests = 500 threads waiting. Memory footprint: ~500MB for 500 threads at 1MB stack each.
- Reactive (non-blocking): Small thread pool (CPU cores * 2). Threads never block. 500 concurrent requests handled on 8 threads using event loops. Memory footprint: much lower.
- Best use case for reactive: I/O-bound microservices doing many parallel calls to databases, external APIs, or message queues. NOT CPU-bound computations.
- Reactive Streams spec (java.util.concurrent.Flow, Java 9+): defines 4 interfaces — Publisher, Subscriber, Subscription, Processor. Project Reactor implements this spec.
- Project Reactor types: Mono<T> = 0 or 1 item (like Optional but async). Flux<T> = 0 to N items (like Stream but async and push-based).
pom.xml (Spring WebFlux)xml
<dependencies>
<!-- Spring WebFlux (reactive HTTP + Reactor Netty) -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!-- R2DBC: reactive database access (PostgreSQL) -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-postgresql</artifactId>
</dependency>
<!-- Testing -->
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>Step 2 — Mono: 0 or 1 Asynchronous Value
Mono<T> is a Publisher that emits at most one item, then completes. It is equivalent to CompletableFuture<Optional<T>> but composable in a reactive pipeline. Nothing happens until you subscribe — reactive pipelines are lazy.
MonoExamples.javajava
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
import java.util.Optional;
public class MonoExamples {
public static void main(String[] args) throws InterruptedException {
// ============================================================
// Creating Monos
// ============================================================
Mono<String> value = Mono.just("Hello"); // emits "Hello" then completes
Mono<String> empty = Mono.empty(); // completes without emitting
Mono<String> error = Mono.error(new RuntimeException("fail")); // emits error signal
Mono<String> defer = Mono.defer(() -> Mono.just("computed-" + System.nanoTime()));
// defer: factory called on EACH subscription — ensures fresh value per subscriber
Mono<String> fromCallable = Mono.fromCallable(() -> {
Thread.sleep(100); // simulate blocking call
return "result from blocking call";
}); // fromCallable wraps a blocking computation — subscribe on boundedElastic!
Mono<String> fromFuture = Mono.fromFuture(() ->
java.util.concurrent.CompletableFuture.supplyAsync(() -> "async result")
);
// ============================================================
// Transforming Monos — operators
// ============================================================
Mono.just(" hello world ")
.map(String::trim) // transform value synchronously
.map(String::toUpperCase) // another synchronous transform
.filter(s -> s.length() > 5) // only continue if predicate matches
.defaultIfEmpty("DEFAULT") // if empty after filter, use this value
.subscribe(
value2 -> System.out.println("Value: " + value2), // onNext
error2 -> System.err.println("Error: " + error2), // onError
() -> System.out.println("Completed") // onComplete
);
// Output: Value: HELLO WORLD
// Completed
// ============================================================
// flatMap: transform into another Mono (async operations)
// Use map() for synchronous transforms, flatMap() for async
// ============================================================
Mono<String> fetchUser = Mono.just("user-1");
Mono<String> fetchOrdersForUser = fetchUser
.flatMap(userId -> {
// This would normally be an async DB/HTTP call
return Mono.just("Orders for " + userId);
});
fetchOrdersForUser.subscribe(System.out::println);
// Output: Orders for user-1
// ============================================================
// zipWith: combine two Monos into a tuple
// Both run in parallel, combine when both emit
// ============================================================
Mono<String> userName = Mono.just("Alice");
Mono<Integer> userScore = Mono.just(95);
Mono.zip(userName, userScore)
.map(tuple -> tuple.getT1() + " scored " + tuple.getT2())
.subscribe(System.out::println);
// Output: Alice scored 95
// ============================================================
// Error handling
// ============================================================
Mono.error(new RuntimeException("service unavailable"))
.onErrorReturn("fallback value") // provide default on any error
.subscribe(System.out::println);
// Output: fallback value
Mono.error(new RuntimeException("timeout"))
.onErrorResume(ex -> { // replace error with another Mono
System.out.println("Error: " + ex.getMessage() + ", trying fallback");
return Mono.just("from cache");
})
.subscribe(System.out::println);
Mono.error(new RuntimeException("fail"))
.onErrorMap(ex -> new IllegalStateException("wrapped: " + ex.getMessage()))
.subscribe(
v -> {},
ex -> System.out.println(ex.getClass().getSimpleName() + ": " + ex.getMessage())
);
// Output: IllegalStateException: wrapped: fail
// ============================================================
// Scheduling: run on different thread pools
// ============================================================
Mono.fromCallable(() -> {
// This blocking call must run on a thread pool that can block
Thread.sleep(100);
return "blocking result";
})
.subscribeOn(Schedulers.boundedElastic()) // run subscription on boundedElastic pool
// Schedulers: immediate(), single(), parallel() (CPU), boundedElastic() (I/O/blocking)
.subscribe(System.out::println);
Thread.sleep(500); // wait for async operations
}
}Step 3 — Flux: 0 to N Asynchronous Items
Flux<T> is a Publisher that emits zero or more items over time, then either completes or errors. It is the reactive equivalent of Iterable or Stream, but items can arrive asynchronously with any timing — from immediate emission to items spaced seconds apart.
FluxExamples.javajava
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.List;
public class FluxExamples {
public static void main(String[] args) throws InterruptedException {
// ============================================================
// Creating Fluxes
// ============================================================
Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5); // from varargs
Flux<String> fromList = Flux.fromIterable(List.of("a", "b", "c")); // from collection
Flux<Integer> range = Flux.range(1, 10); // 1 to 10
Flux<Long> interval = Flux.interval(Duration.ofMillis(100)); // 0,1,2,... every 100ms
Flux<Integer> concat = Flux.concat(Flux.just(1,2), Flux.just(3,4)); // sequential
Flux<Integer> merge = Flux.merge(Flux.just(1,2), Flux.just(3,4)); // concurrent
// ============================================================
// Essential operators — mimic Stream API but async
// ============================================================
Flux.range(1, 10)
.filter(n -> n % 2 == 0) // 2, 4, 6, 8, 10
.map(n -> n * n) // 4, 16, 36, 64, 100
.take(3) // 4, 16, 36 — stop after 3
.subscribe(n -> System.out.print(n + " "));
System.out.println(); // 4 16 36
// ============================================================
// flatMap: async transform, results may interleave
// concatMap: sequential async transform, preserves order
// ============================================================
Flux.just("user-1", "user-2", "user-3")
.flatMap(userId -> fetchOrders(userId)) // all 3 fetched in parallel
.subscribe(order -> System.out.println("Got order: " + order));
Flux.just("user-1", "user-2", "user-3")
.concatMap(userId -> fetchOrders(userId)) // sequential: user-1 done, then user-2, ...
.subscribe(order -> System.out.println("Sequential order: " + order));
Thread.sleep(500);
// ============================================================
// Collecting Flux items
// ============================================================
Mono<List<Integer>> list = Flux.range(1, 5).collectList(); // Flux -> Mono<List>
list.subscribe(l -> System.out.println("List: " + l));
Mono<Long> count = Flux.range(1, 100).count();
count.subscribe(c -> System.out.println("Count: " + c));
Mono<Integer> sum = Flux.range(1, 10)
.reduce(0, Integer::sum); // reduce like Stream.reduce
sum.subscribe(s -> System.out.println("Sum: " + s));
// collectMap: Flux -> Mono<Map>
Mono<java.util.Map<Integer, Integer>> squaresMap = Flux.range(1, 5)
.collectMap(n -> n, n -> n * n);
squaresMap.subscribe(m -> System.out.println("Map: " + m));
// ============================================================
// Windowing and batching
// ============================================================
Flux.range(1, 10)
.buffer(3) // emit List<Integer> of up to 3 items: [1,2,3], [4,5,6], [7,8,9], [10]
.subscribe(batch -> System.out.println("Batch: " + batch));
Flux.range(1, 10)
.window(3) // emit inner Flux of up to 3 items (streaming, not buffered)
.flatMap(window -> window.collectList())
.subscribe(w -> System.out.println("Window: " + w));
Thread.sleep(200);
// ============================================================
// Grouping
// ============================================================
Flux.range(1, 10)
.groupBy(n -> n % 2 == 0 ? "even" : "odd")
.flatMap(group -> group.collectList().map(items -> group.key() + ": " + items))
.subscribe(System.out::println);
Thread.sleep(200);
}
// Simulates async I/O that returns multiple items per user
static Flux<String> fetchOrders(String userId) {
return Flux.just(userId + "-order-1", userId + "-order-2")
.delayElements(Duration.ofMillis(50));
}
}Step 4 — Backpressure: Controlling Flow
Backpressure is the mechanism by which a slow consumer signals to a fast producer to slow down. Without backpressure, a producer emitting faster than the consumer processes would overflow a buffer and eventually cause out-of-memory errors. Reactor handles backpressure through the Reactive Streams request protocol.
BackpressureExample.javajava
import reactor.core.publisher.Flux;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
public class BackpressureExample {
public static void main(String[] args) throws InterruptedException {
// ============================================================
// Custom subscriber with backpressure control
// ============================================================
Flux.range(1, 100)
.subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(reactor.core.CoreSubscription subscription) {
// Request first 5 items (pull-based start)
request(5);
}
int count = 0;
@Override
protected void hookOnNext(Integer value) {
System.out.println("Processing: " + value);
count++;
if (count % 5 == 0) {
// After every 5 items, request 5 more
request(5);
}
// Never call request(Long.MAX_VALUE) unless you can handle all items
}
@Override
protected void hookOnComplete() {
System.out.println("Done! Processed " + count + " items");
}
});
Thread.sleep(500);
// ============================================================
// Overflow strategies for mismatched producer/consumer speed
// ============================================================
// onBackpressureBuffer: buffer all items (can OOM if unlimited!)
Flux.interval(Duration.ofMillis(1)) // fast producer: 1000/s
.onBackpressureBuffer(100) // buffer up to 100, then error
.publishOn(Schedulers.parallel())
.subscribe(i -> {
try { Thread.sleep(10); } // slow consumer: 100/s
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
// System.out.println(i);
});
Thread.sleep(200);
// onBackpressureDrop: drop items when consumer is not ready
Flux.interval(Duration.ofMillis(1))
.onBackpressureDrop(dropped -> System.out.println("Dropped: " + dropped))
.publishOn(Schedulers.parallel())
.take(20)
.subscribe(i -> {
try { Thread.sleep(10); }
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
});
Thread.sleep(500);
// onBackpressureLatest: keep only the most recent unprocessed item
Flux.interval(Duration.ofMillis(1))
.onBackpressureLatest()
.publishOn(Schedulers.parallel())
.take(20)
.subscribe(i -> {
try { Thread.sleep(10); }
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
});
Thread.sleep(500);
// onBackpressureError: throw OverflowException when buffer fills
System.out.println("Backpressure demo complete");
}
}Step 5 — Combining and Merging Streams
Reactive programming shines when combining multiple async sources. Reactor provides operators for merging, zipping, concatenating, and switching between Fluxes in controlled ways.
CombiningFlux.javajava
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
public class CombiningFlux {
public static void main(String[] args) throws InterruptedException {
// ============================================================
// Flux.merge: subscribe to ALL sources simultaneously
// Items from all sources interleave as they arrive
// ============================================================
Flux<String> source1 = Flux.just("A", "B", "C")
.delayElements(Duration.ofMillis(100));
Flux<String> source2 = Flux.just("1", "2", "3")
.delayElements(Duration.ofMillis(150));
Flux.merge(source1, source2) // concurrent — order not guaranteed
.subscribe(item -> System.out.print(item + " "));
Thread.sleep(600);
System.out.println(); // A 1 B 2 C 3 (approximate — interleaved by timing)
// ============================================================
// Flux.concat: subscribe to sources ONE AT A TIME, preserving order
// source2 starts only after source1 completes
// ============================================================
Flux.concat(source1, source2)
.subscribe(item -> System.out.print(item + " "));
Thread.sleep(1000);
System.out.println(); // A B C 1 2 3
// ============================================================
// Flux.zip: pair items from N sources by position
// Emits only when ALL sources have emitted their Nth item
// ============================================================
Flux<String> names = Flux.just("Alice", "Bob", "Charlie");
Flux<Integer> ages = Flux.just(30, 25, 35);
Flux<String> cities = Flux.just("NYC", "LA", "Chicago");
Flux.zip(names, ages, cities)
.map(tuple -> tuple.getT1() + ", " + tuple.getT2() + ", " + tuple.getT3())
.subscribe(System.out::println);
Thread.sleep(100);
// Alice, 30, NYC
// Bob, 25, LA
// Charlie, 35, Chicago
// Mono.zip: combine results from multiple Monos
Mono<String> userName = Mono.just("Alice").delayElement(Duration.ofMillis(100));
Mono<Integer> score = Mono.just(95).delayElement(Duration.ofMillis(50));
Mono.zip(userName, score)
.map(t -> t.getT1() + " scored " + t.getT2())
.subscribe(System.out::println);
Thread.sleep(200);
// Alice scored 95
// ============================================================
// switchMap: for each item, switch to a NEW inner flux, cancelling the old one
// Classic use: autocomplete — only care about the LATEST search term
// ============================================================
Flux.just("J", "Ja", "Jav", "Java") // simulates typing in a search box
.delayElements(Duration.ofMillis(50))
.switchMap(term -> {
System.out.println("Searching for: " + term);
return Flux.just("result for " + term)
.delayElements(Duration.ofMillis(80));
// Previous search is cancelled when new term arrives
})
.subscribe(result -> System.out.println("Got: " + result));
Thread.sleep(600);
// Only "result for Java" will be emitted — previous searches are cancelled
// ============================================================
// combineLatest: emit whenever ANY source emits, with latest from all others
// Used for reactive forms (validate when any field changes)
// ============================================================
Flux<String> field1 = Flux.just("alice", "alice@", "alice@example.com")
.delayElements(Duration.ofMillis(100));
Flux<String> field2 = Flux.just("pass", "pass123", "pass123!")
.delayElements(Duration.ofMillis(150));
Flux.combineLatest(field1, field2,
(email, pwd) -> "email=" + email + " pwd=" + pwd)
.subscribe(System.out::println);
Thread.sleep(600);
}
}Step 6 — Spring WebFlux: Reactive REST Controllers
Spring WebFlux is the reactive web framework built on Project Reactor and Reactor Netty. It supports the same @Controller/@RestController annotation model as Spring MVC, but handler methods return Mono or Flux instead of plain objects. WebFlux runs on a small event loop — typically one thread per CPU core — and handles all I/O asynchronously.
ReactiveUserController.javajava
import org.springframework.http.*;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.*;
import java.net.URI;
import java.time.Duration;
import jakarta.validation.Valid;
@RestController
@RequestMapping("/api/users")
public class ReactiveUserController {
private final ReactiveUserService userService;
private final ReactiveUserRepository userRepository;
public ReactiveUserController(ReactiveUserService service, ReactiveUserRepository repo) {
this.userService = service;
this.userRepository = repo;
}
// GET /api/users — stream all users
// Flux<UserDTO> returned directly — Spring serializes to JSON array
@GetMapping(produces = MediaType.APPLICATION_JSON_VALUE)
public Flux<UserDTO> getAllUsers(
@RequestParam(defaultValue = "0") int page,
@RequestParam(defaultValue = "20") int size) {
return userRepository.findAll()
.skip((long) page * size)
.take(size)
.map(UserDTO::fromEntity);
}
// GET /api/users/{id} — returns Mono<ResponseEntity>
@GetMapping("/{id}")
public Mono<ResponseEntity<UserDTO>> getUser(@PathVariable Long id) {
return userService.findById(id)
.map(user -> ResponseEntity.ok(UserDTO.fromEntity(user)))
.defaultIfEmpty(ResponseEntity.notFound().build());
// If Mono is empty, return 404. No if/else — reactive operators handle it.
}
// POST /api/users — create and return 201 Created
@PostMapping
public Mono<ResponseEntity<UserDTO>> createUser(@Valid @RequestBody CreateUserRequest req) {
return userService.createUser(req.name(), req.email())
.map(saved -> ResponseEntity
.created(URI.create("/api/users/" + saved.getId()))
.body(UserDTO.fromEntity(saved)));
}
// DELETE /api/users/{id}
@DeleteMapping("/{id}")
public Mono<ResponseEntity<Void>> deleteUser(@PathVariable Long id) {
return userService.deleteById(id)
.then(Mono.just(ResponseEntity.<Void>noContent().build()));
// then(): discard value, emit a new signal when upstream completes
}
// ============================================================
// Server-Sent Events (SSE): real-time streaming to browser
// ============================================================
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<UserDTO> streamUsers() {
// Emits a new user every second — browser receives as SSE events
return Flux.interval(Duration.ofSeconds(1))
.flatMap(tick -> userRepository.findAll().take(1))
.map(UserDTO::fromEntity)
.take(Duration.ofMinutes(5)); // stop streaming after 5 minutes
}
// ============================================================
// Reactive request body with streaming JSON
// ============================================================
@PostMapping(value = "/batch",
consumes = MediaType.APPLICATION_NDJSON_VALUE, // newline-delimited JSON
produces = MediaType.APPLICATION_JSON_VALUE)
public Mono<Long> createBatch(@RequestBody Flux<CreateUserRequest> requests) {
return requests
.flatMap(req -> userService.createUser(req.name(), req.email()))
.count(); // returns count of successfully created users
}
// Placeholder types
record CreateUserRequest(String name, String email) {}
record UserDTO(Long id, String name, String email) {
static UserDTO fromEntity(Object entity) { return new UserDTO(1L, "name", "email"); }
}
}Step 7 — Functional Routing in WebFlux
WebFlux offers a second programming model: functional routing. Instead of @Controller annotations, you define routes using RouterFunction and handle requests with HandlerFunction. This is more explicit and easier to test in isolation — the router and handler are plain objects, not Spring beans with magic annotations.
FunctionalRouting.javajava
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.*;
import reactor.core.publisher.Mono;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;
import static org.springframework.web.reactive.function.server.RequestPredicates.*;
// ============================================================
// Handler: processes requests, returns ServerResponse
// ============================================================
@Component
class UserHandler {
private final ReactiveUserService userService;
UserHandler(ReactiveUserService userService) {
this.userService = userService;
}
public Mono<ServerResponse> getUser(ServerRequest request) {
Long id = Long.parseLong(request.pathVariable("id"));
return userService.findById(id)
.flatMap(user -> ServerResponse.ok()
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(user))
.switchIfEmpty(ServerResponse.notFound().build());
}
public Mono<ServerResponse> getAllUsers(ServerRequest request) {
return ServerResponse.ok()
.contentType(MediaType.APPLICATION_JSON)
.body(userService.findAll(), UserEntity.class);
}
public Mono<ServerResponse> createUser(ServerRequest request) {
return request.bodyToMono(CreateUserRequest.class)
.flatMap(req -> userService.createUser(req.name(), req.email()))
.flatMap(user -> ServerResponse.created(
java.net.URI.create("/api/users/" + user.getId()))
.bodyValue(user));
}
record CreateUserRequest(String name, String email) {}
}
// ============================================================
// Router: maps URL patterns to handler methods
// ============================================================
@Configuration
class UserRouter {
@Bean
RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
return route()
.GET("/api/users", accept(MediaType.APPLICATION_JSON), handler::getAllUsers)
.GET("/api/users/{id}", accept(MediaType.APPLICATION_JSON), handler::getUser)
.POST("/api/users", contentType(MediaType.APPLICATION_JSON), handler::createUser)
.build();
}
// Nested routes with path prefix
@Bean
RouterFunction<ServerResponse> nestedRoutes(UserHandler handler) {
return route()
.path("/api", builder -> builder
.path("/users", inner -> inner
.GET("", handler::getAllUsers)
.GET("/{id}", handler::getUser)
.POST("", handler::createUser)
)
)
.build();
}
}Step 8 — WebClient: Non-Blocking HTTP Client
WebClient is the reactive replacement for RestTemplate. It is completely non-blocking and returns Mono/Flux — the response is processed in the reactive pipeline without occupying a thread while waiting for the HTTP response.
WebClientExample.javajava
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.*;
import org.springframework.http.MediaType;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import reactor.util.retry.Retry;
import java.time.Duration;
@Service
public class WebClientExample {
private final WebClient webClient;
public WebClientExample(WebClient.Builder builder) {
this.webClient = builder
.baseUrl("https://api.example.com")
.defaultHeader("Content-Type", "application/json")
.defaultHeader("Accept", "application/json")
.codecs(c -> c.defaultCodecs().maxInMemorySize(1024 * 1024)) // 1MB max body
.build();
}
// ============================================================
// GET: retrieve a single resource
// ============================================================
public Mono<UserDTO> getUser(Long userId) {
return webClient.get()
.uri("/users/{id}", userId)
.retrieve()
// retrieve() checks HTTP status — 4xx/5xx become Mono.error automatically
.onStatus(status -> status.is4xxClientError(),
response -> response.bodyToMono(ErrorResponse.class)
.map(err -> new RuntimeException("Client error: " + err.message())))
.onStatus(status -> status.is5xxServerError(),
response -> Mono.error(new RuntimeException("Server error: " + response.statusCode())))
.bodyToMono(UserDTO.class); // deserialize JSON body to UserDTO
}
// ============================================================
// GET: retrieve a list as Flux (streaming)
// ============================================================
public Flux<UserDTO> getAllUsers() {
return webClient.get()
.uri("/users")
.retrieve()
.bodyToFlux(UserDTO.class); // stream JSON array as Flux
}
// ============================================================
// POST: send a request body
// ============================================================
public Mono<UserDTO> createUser(String name, String email) {
record CreateRequest(String name, String email) {}
return webClient.post()
.uri("/users")
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(new CreateRequest(name, email)) // serialize to JSON
.retrieve()
.bodyToMono(UserDTO.class);
}
// ============================================================
// Retry on transient failures
// ============================================================
public Mono<UserDTO> getUserWithRetry(Long userId) {
return webClient.get()
.uri("/users/{id}", userId)
.retrieve()
.bodyToMono(UserDTO.class)
.retryWhen(Retry.backoff(3, Duration.ofMillis(100)) // 3 retries, exponential backoff
.maxBackoff(Duration.ofSeconds(2))
.filter(ex -> ex instanceof WebClientRequestException) // only retry on connection errors
)
.timeout(Duration.ofSeconds(5))
.onErrorReturn(new UserDTO(0L, "unknown", ""));
}
// ============================================================
// Parallel calls with Mono.zip
// ============================================================
public Mono<String> getUserAndOrders(Long userId) {
Mono<UserDTO> user = getUser(userId);
Mono<String> orders = webClient.get()
.uri("/users/{id}/orders", userId)
.retrieve()
.bodyToMono(String.class);
// Both run in parallel; combine when both complete
return Mono.zip(user, orders)
.map(tuple -> "User: " + tuple.getT1().name() + ", Orders: " + tuple.getT2());
}
// ============================================================
// Exchange: low-level access to the full response
// ============================================================
public Mono<String> headersAndBody(Long userId) {
return webClient.get()
.uri("/users/{id}", userId)
.exchangeToMono(response -> { // exchangeToMono: you must handle ALL status codes
System.out.println("Status: " + response.statusCode());
System.out.println("Headers: " + response.headers().asHttpHeaders());
if (response.statusCode().is2xxSuccessful()) {
return response.bodyToMono(String.class);
} else {
return response.createError(); // turn non-2xx into Mono.error
}
});
}
// Placeholder types
record UserDTO(Long id, String name, String email) {}
record ErrorResponse(String message) {}
}Step 9 — R2DBC: Reactive Database Access
R2DBC (Reactive Relational Database Connectivity) is the reactive alternative to JDBC. JDBC is inherently blocking — it was designed for one-thread-per-query. R2DBC uses non-blocking database drivers, returning Flux/Mono from every query so your reactive pipeline stays non-blocking all the way to the database.
application.yml (R2DBC)yaml
spring:
r2dbc:
url: r2dbc:postgresql://localhost:5432/myapp
username: appuser
password: secret
pool:
enabled: true
initial-size: 5
max-size: 20
max-idle-time: 30m
# R2DBC doesn't use spring.datasource — that's for JDBC
# Use Flyway/Liquibase with a JDBC datasource for schema migrations
# (migrations run before app start, then R2DBC handles queries)
logging:
level:
io.r2dbc.postgresql: DEBUG # log R2DBC SQL queriesReactiveUserRepository.javajava
import org.springframework.data.annotation.*;
import org.springframework.data.relational.core.mapping.*;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.domain.*;
import java.time.LocalDateTime;
// ============================================================
// R2DBC Entity: use @Table, @Id from Spring Data Relational
// NOT JPA annotations — R2DBC is NOT JPA
// ============================================================
@Table("users")
class UserEntity {
@Id
private Long id;
private String name;
private String email;
private boolean active;
@Column("created_at")
private LocalDateTime createdAt;
// getters and setters...
}
// ============================================================
// Reactive Repository: all methods return Mono or Flux
// ============================================================
interface ReactiveUserRepository extends ReactiveCrudRepository<UserEntity, Long> {
// Derived queries: same naming as Spring Data JPA
Flux<UserEntity> findByActiveTrue(); // SELECT * FROM users WHERE active = true
Mono<UserEntity> findByEmail(String email); // SELECT * FROM users WHERE email = ?
Mono<Boolean> existsByEmail(String email);
Mono<Long> countByActiveTrue();
// Sorted result
Flux<UserEntity> findByActiveTrueOrderByNameAsc();
// Custom @Query with R2DBC SQL syntax
@Query("SELECT * FROM users WHERE name ILIKE :pattern ORDER BY name LIMIT :limit")
Flux<UserEntity> searchByName(String pattern, int limit);
// Modifying query (returns Mono<Integer> = affected rows)
@org.springframework.data.r2dbc.repository.Modifying
@Query("UPDATE users SET active = :active WHERE id = :id")
Mono<Integer> updateActiveStatus(Long id, boolean active);
@org.springframework.data.r2dbc.repository.Modifying
@Query("DELETE FROM users WHERE active = false AND created_at < :cutoff")
Mono<Integer> deleteInactiveUsersBefore(LocalDateTime cutoff);
}
// ============================================================
// Service: uses the reactive repository in reactive pipelines
// ============================================================
class ReactiveUserService {
private final ReactiveUserRepository userRepo;
ReactiveUserService(ReactiveUserRepository userRepo) {
this.userRepo = userRepo;
}
public Mono<UserEntity> createUser(String name, String email) {
return userRepo.existsByEmail(email)
.flatMap(exists -> {
if (exists) {
return Mono.error(new IllegalStateException("Email already registered: " + email));
}
UserEntity user = new UserEntity();
user.setName(name);
user.setEmail(email);
user.setActive(true);
user.setCreatedAt(LocalDateTime.now());
return userRepo.save(user);
});
}
public Mono<UserEntity> findById(Long id) {
return userRepo.findById(id);
}
public Flux<UserEntity> findAll() {
return userRepo.findAll();
}
public Mono<Void> deleteById(Long id) {
return userRepo.deleteById(id);
}
// ============================================================
// R2DBC transactions with @Transactional
// Works reactively — transaction spans the Mono/Flux pipeline
// ============================================================
@org.springframework.transaction.annotation.Transactional
public Mono<UserEntity> transferAndUpdate(Long fromId, Long toId) {
return userRepo.findById(fromId)
.zipWith(userRepo.findById(toId))
.flatMap(tuple -> {
UserEntity from = tuple.getT1();
UserEntity to = tuple.getT2();
from.setActive(false);
to.setName(to.getName() + "(promoted)");
return userRepo.save(from)
.then(userRepo.save(to)); // both in same transaction
});
// If either save fails, the entire transaction rolls back
}
}Step 10 — Testing Reactive Code with StepVerifier
StepVerifier is the testing tool from reactor-test. It subscribes to a Mono or Flux and lets you assert on each emitted item, the completion signal, and errors — all in a time-controlled, synchronous test method. You don't need Thread.sleep() to test reactive code.
ReactiveTests.javajava
import org.junit.jupiter.api.*;
import reactor.core.publisher.*;
import reactor.test.StepVerifier;
import reactor.test.scheduler.VirtualTimeScheduler;
import java.time.Duration;
class ReactiveTests {
// ============================================================
// Testing Mono
// ============================================================
@Test
void monoEmitsValue() {
Mono<String> mono = Mono.just("hello");
StepVerifier.create(mono)
.expectNext("hello") // assert next emitted value
.verifyComplete(); // assert the Mono completes (no error)
}
@Test
void monoIsEmpty() {
StepVerifier.create(Mono.empty())
.verifyComplete(); // empty Mono completes without emitting
}
@Test
void monoErrors() {
Mono<String> error = Mono.error(new RuntimeException("oops"));
StepVerifier.create(error)
.expectErrorMessage("oops") // assert error message
.verify(); // don't use verifyComplete() on error paths
// Or check error type:
StepVerifier.create(Mono.error(new IllegalArgumentException("bad")))
.expectError(IllegalArgumentException.class)
.verify();
}
// ============================================================
// Testing Flux
// ============================================================
@Test
void fluxEmitsInOrder() {
Flux<Integer> flux = Flux.range(1, 5);
StepVerifier.create(flux)
.expectNext(1)
.expectNext(2)
.expectNext(3)
.expectNext(4)
.expectNext(5)
.verifyComplete();
// Shorthand for multiple items:
StepVerifier.create(Flux.range(1, 5))
.expectNext(1, 2, 3, 4, 5)
.verifyComplete();
}
@Test
void fluxFiltered() {
Flux<Integer> evens = Flux.range(1, 10).filter(n -> n % 2 == 0);
StepVerifier.create(evens)
.expectNext(2, 4, 6, 8, 10)
.verifyComplete();
}
@Test
void assertWithPredicate() {
StepVerifier.create(Mono.just("Hello World"))
.expectNextMatches(s -> s.startsWith("Hello") && s.length() == 11)
.verifyComplete();
}
@Test
void consumeAndAssert() {
StepVerifier.create(Flux.just("A", "B", "C"))
.expectNextCount(2) // consume next 2 items without checking values
.expectNext("C")
.verifyComplete();
}
// ============================================================
// Testing time-dependent reactive streams with VirtualTime
// Without VirtualTime, a 5-second interval test would take 5 real seconds
// ============================================================
@Test
void testWithVirtualTime() {
// MUST use withVirtualTime() for any time-based operators
StepVerifier.withVirtualTime(() ->
Flux.interval(Duration.ofSeconds(1)).take(3)
)
.thenAwait(Duration.ofSeconds(3)) // advance virtual clock 3 seconds
.expectNext(0L, 1L, 2L)
.verifyComplete();
// This test runs in MILLISECONDS, not 3 real seconds
}
@Test
void testDelayElement() {
StepVerifier.withVirtualTime(() ->
Mono.just("delayed").delayElement(Duration.ofMinutes(5))
)
.thenAwait(Duration.ofMinutes(5)) // fast-forward 5 minutes
.expectNext("delayed")
.verifyComplete();
// Completes instantly in test
}
// ============================================================
// Testing error recovery
// ============================================================
@Test
void testOnErrorReturn() {
Mono<String> withFallback = Mono.error(new RuntimeException("fail"))
.onErrorReturn("fallback");
StepVerifier.create(withFallback)
.expectNext("fallback")
.verifyComplete();
}
// ============================================================
// Testing reactive service
// ============================================================
@Test
void testReactiveService() {
// Mock the repository
ReactiveUserRepository mockRepo = org.mockito.Mockito.mock(ReactiveUserRepository.class);
org.mockito.Mockito.when(mockRepo.existsByEmail("new@test.com")).thenReturn(Mono.just(false));
org.mockito.Mockito.when(mockRepo.save(org.mockito.ArgumentMatchers.any()))
.thenAnswer(inv -> Mono.just(inv.getArgument(0)));
ReactiveUserService service = new ReactiveUserService(mockRepo);
StepVerifier.create(service.createUser("Test", "new@test.com"))
.expectNextMatches(user -> user.getName().equals("Test"))
.verifyComplete();
}
}Step 11 — Common Pitfalls in Reactive Programming
Reactive Anti-Patterns to Avoid
- Never call block() inside a reactive pipeline. block() bridges reactive to blocking and will deadlock when called on the Reactor Netty event loop thread. Only call block() at the outermost level in tests or main() methods.
- Don't mix reactive and blocking code on the event loop. If you must call a blocking API (legacy JDBC, file I/O), wrap it in Mono.fromCallable(...).subscribeOn(Schedulers.boundedElastic()) to offload it to a blocking-capable thread pool.
- Reactive pipelines don't run until subscribed. Mono.just(saveToDatabase()) executes saveToDatabase() immediately (eagerly) because it's an argument expression. Use Mono.defer(() -> Mono.just(saveToDatabase())) or Mono.fromCallable(() -> saveToDatabase()) for lazy evaluation.
- Don't use ThreadLocal in reactive code. Each reactive operator may run on a different thread, so ThreadLocal values set in one operator are not visible in the next. Use Reactor Context (context().put/get) for propagating per-request state through a reactive pipeline.
- Avoid flatMap with unbounded concurrency on slow services. Flux.flatMap() subscribes to ALL inner publishers simultaneously by default. If the outer Flux emits 10,000 items and each flatMap makes an HTTP call, you make 10,000 concurrent requests. Use flatMap(mapper, concurrency) to limit: .flatMap(item -> call(item), 100) for 100 concurrent calls max.
- Always add .timeout() on external calls. A WebClient call without a timeout will hang forever if the server never responds, holding resources. Always add .timeout(Duration.ofSeconds(5)) before subscribing.
- Use Flux.error() not throw for errors in pipelines. Throwing an exception inside a reactive operator like map() is usually caught and turned into an error signal. But it's fragile. Return Mono.error() or Flux.error() explicitly for clean error propagation.