Java Reactive Programming

Table of Contents

1. Introduction and course overview
2. The Reactive Revolution

• 2.1 What is reactive programming?
• 2.2 Why adopt reactive programming?
• 2.3 Reactive programming vs reactive systems
• 2.4 The backpressure
• 2.5 The building blocks — Publisher, Subscriber, Subscription
• 2.6 Memory efficiency

3. Scaling with Reactive Microservices

• 3.1 Setting up the Spring Boot + WebFlux project
• 3.2 Backpressure en code — BackPressureOne
• 3.3 Reactive Controllers with Mono and Flux
• 3.4 Spring WebFlux internal architecture
• 3.5 Tests with WebTestClient
• 3.6 Reactive stream tests with StepVerifier
• 3.7 Reactive operators — the pipeline
• 3.8 Data streaming with Server-Sent Events (SSE)
• 3.9 Alternative — Les Virtual Threads

4. Demonstration Project Reference

• 4.1 pom.xml — Maven dependencies
• 4.2 Source structure
• 4.3 Model Classes
• 4.4 Services
• 4.5 Components and listeners
• 4.6 Unit and integration tests
• 4.7 Advanced example — GnarlyReactiveOrderProcessor

5. Versions and prerequisites
6. Summary of key concepts

1. Course Introduction and Overview

This course is an accelerated introduction to reactive programming with Java, Spring WebFlux and Project Reactor. It allows you to build modern, high-performance applications that can scale easily and respond in milliseconds.

At the end of the course, you will be able to:

• Write clean, efficient, non-blocking Java code using the reactive model.
• Explore Project Reactor and its main types: Mono and Flux.
• Building Reactive APIs with Spring WebFlux.
• Connect databases without blocking threads.
• Explain how your application uses fewer resources to do more work.
• Understand the concept of backpressure and its implementation.
• Test responsive streams with WebTestClient and StepVerifier.
• Understand the emerging alternative offered by the JDK’s virtual threads.

Versions used in this course: Spring Boot 3.5.0 · Project Reactor 3.7.6 · Spring WebFlux 6.1.14 · Java JDK 24+

In the final segment, the course also presents a new reactive approach that does not use any reactive framework, relying solely on the latest features of the JDK — some finalized, others still in preview.

2. The Reactive Revolution

2.1 What is reactive programming?

Reactive programming is a paradigm, a way of thinking about and writing code that reacts to data, events, and signals. Instead of pulling data or blocking threads, we listen to the data, respond asynchronously, and process everything as a stream.

It is particularly well suited to I/O-bound applications:

• APIs that wait for database responses.
• Microservices that call other services.
• Web applications managing thousands of concurrent users.

Reactive programming doesn’t revolve around threads — it revolves around managing data flow efficiently without wasting resources.

Comparing query processing models

In the fully reactive model, instead of blocking a thread while waiting for a response, we release it. This thread will process other requests, then when the response arrives, the system notifies us and resumes processing asynchronously. This is where all the gain lies.

2.2 Why adopt reactive programming?

Here are the concrete benefits:

1. Better CPU utilization — Threads stay busy doing real work, not stuck waiting.
2. Higher throughput — More requests per second, especially under I/O pressure.
3. Reduced cloud costs — Handle more load with fewer instances = real savings.
4. Graceful scalability — No thread pool explosion when users increase.

You don’t change what your application does — you optimize how it does it.

Spring WebFlux vs Spring MVC — Compared Overview

Note: Netty, Undertow and Jetty are containers that work well with Spring WebFlux to enable the reactive processing model. Tomcat can be used, but only in non-blocking mode with Servlet 3.1 or higher, the subscription still being managed by Spring WebFlux.

Real use case: Fintech — Payment Status API

Imagine a fintech company that processes real-time payments between merchants and customers. A central service is a Payment Status API which returns the current status of a transaction.

Under load, synchronous REST endpoints with blocking I/O begin to show limitations, especially during high volume periods like Black Friday or tax deadlines. The system slows down, loses queries, or requires costly horizontal scaling.

This is where reactive programming comes in: by handling many concurrent requests without blocking threads, we maintain low latency and happy customers, even under pressure.

2.3 Reactive Programming vs. Reactive Systems

This is a common point of confusion:

Reactive programming is a building block of reactive systems. But as soon as you introduce external components — like a message broker or a second process — you enter reactive systems territory.

Example: available memory exceeded with a broker message

If you need to buffer more than memory can hold, you can offload messages to a broker like ActiveMQ. The subscriber can then pull messages only when he is ready. But as soon as you do that — even if ActiveMQ runs on the same VM but in a separate process — you’ve crossed the line into reactive systems. You must then manage:

• Serialization.
• fault tolerance.
• Delivery guarantees.

2.4 Backpressure

One of the smartest features of reactive programming is how it handles backpressure.

Imagine this: a subscriber can say to the publisher: “Hey, slow down, I’m not ready yet.”

This is controlled by calling subscription.request(n). The subscriber requests exactly how many items it can handle via the method parameter — the publisher must obey.

If the publisher produces too quickly, you can:

• Buffer elements.
• Abandon them (drop).
• Raise an error (error).

This is all configurable.

Backpressure mechanism diagram

Publisher (Flux.range 1-100)
    │
    │  subscription.request(5)  ◄────────────────────────┐
    │                                                      │
    ▼                                                      │
  limitRate(5)                                     Subscriber
    │                                                      │
    │  onNext(item1..5)  ─────────────────────────────►   │
    │                                               traitement
    │                                                      │
    │  subscription.request(5)  ◄──────────────── batch_size=5
    │
   [...]
    │
    │  onComplete()  ──────────────────────────────────►  │

2.5 Building Blocks — Publisher, Subscriber, Subscription

The fundamental interfaces are in the reactive-streams jar:

The Processor is rarely used directly, but plays a crucial role in advanced reactive pipelines.

Publisher–Subscriber interaction sequence

Subscriber                         Publisher
    │                                  │
    │── subscribe() ─────────────────► │
    │                                  │
    │ ◄─── onSubscribe(subscription) ──│
    │                                  │
    │── subscription.request(n) ─────► │
    │                                  │
    │ ◄─── onNext(item1) ──────────────│
    │ ◄─── onNext(item2) ──────────────│
    │ ◄─── ...                         │
    │ ◄─── onNext(itemN) ──────────────│
    │                                  │
    │ ◄─── onComplete() ───────────────│  (ou onError() en cas d'erreur)

Sequence step by step:

1. Subscription starts when the Subscriber calls the subscribe method on the Publisher, saying in essence: “I want to receive data.”
2. The Publisher responds by creating a Subscription object and sending it to the Subscriber: “You are now subscribed.”
3. The Subscriber calls request(n) on the Subscription, telling the Publisher: “Send me n items.”
4. The Publisher sends the actual data via the onNext(data) method, called n times depending on the number requested.
5. After sending all the data, the Publisher calls onComplete() — “I’m done.”
6. In case of an error, the Publisher calls onError(throwable) instead of onComplete().

2.6 Memory efficiency

Do not neglect RAM usage:

• Traditional applications — Create hundreds of threads, each with its own memory stack (~1MB each). It adds up quickly.
• Reactive applications — Need only a few threads, often one per CPU core. Fewer threads = less memory = more room for actual data and business logic.

3. Scaling with Reactive Microservices

3.1 Setting up the Spring Boot + WebFlux project

Via Spring Initializr

The fastest way to create a responsive Spring Boot project is to use Spring Initializr (start.spring.io). Stable selections for this course:

The WebFlux starter automatically pulls reactor-core, so there is no need to add Reactor dependencies manually.

basic pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

This single dependency is enough to get started with Spring WebFlux and Project Reactor.

3.2 Backpressure in code — BackPressureOne

Before diving into controllers, here is a concrete example that illustrates the fundamental reactive interfaces: Publisher, Subscriber and Subscription.

In this example, we create a Flux (a Publisher) which emits integers from 1 to 100. We apply limitRate(5) to never send more than 5 elements per downstream request. Then, we subscribe with an anonymous implementation of Subscriber<Integer>.

The crucial point is in onSubscribe: we call subscription.request(BATCH_SIZE) with BATCH_SIZE = 5, thus requesting 5 elements from the Publisher. In onNext, once 5 elements have been received, a new batch is requested.

package com.example.demo;

import reactor.core.publisher.Flux;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class BackPressureOne {

    public static void main(String[] args) {

        Flux<Integer> flux = Flux.range(1, 100)
                .doOnRequest(n -> System.out.println("[Flux] Request received: " + n))
                .limitRate(5); // Limiter le débit à 5 éléments par requête downstream

        flux.subscribe(new Subscriber<Integer>() {

            private Subscription subscription;
            private int count = 0;
            private final int BATCH_SIZE = 5;

            @Override
            public void onSubscribe(Subscription s) {
                this.subscription = s;
                System.out.println("[Subscriber] Subscribed");
                subscription.request(BATCH_SIZE); // Demande initiale de 5 éléments
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("[Subscriber] Received: " + item);
                count++;
                if (count % BATCH_SIZE == 0) {
                    System.out.println("[Subscriber] Requesting next batch of " + BATCH_SIZE);
                    subscription.request(BATCH_SIZE); // Demande le prochain batch
                }
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("[Subscriber] Error: " + t.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("[Subscriber] Completed");
            }
        });
    }
}

Key points from this example:

• Stream.range(1, 100) — Creates a Publisher that emits 100 integers.
• .doOnRequest(n -> ...) — Side effect to visualize backpressure requests.
• .limitRate(5) — Imposes a limit of 5 elements maximum per downstream request.
• subscription.request(BATCH_SIZE) — Explicit flow control, the Subscriber dictates its pace.

3.3 Responsive Controllers with Mono and Flux

Understanding reactive types returned by controllers

Let’s create a simple @RestController. The method returns Mono<String> — a reactive type that will optionally emit a single value of type String.

package com.example.demo.controller;

import com.example.demo.operators.StockQuote;
import org.jetbrains.annotations.NotNull;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;

@RestController
@RequestMapping("/api/v1")
public class ReactionControllerOne {

    @GetMapping("/message")
    public Mono<String> getMessage() {
        // On retourne un Mono qui émet un seul message
        return Mono.just("Sorry, couldn't help, but react!").log();
    }

    @GetMapping("/demo-pipeline")
    public Flux<StockQuote> demoPipeline() {
        return getQuotesForIndustry("software")
                .map(quote -> {
                    quote.setLanguageCode("ENRICHED");
                    return quote;
                })
                .filter(quote -> quote.getLanguageCode().startsWith("E"))
                .take(1);
    }

    private @NotNull Flux<StockQuote> getQuotesForIndustry(String industryCode) {
        return Flux.range(1, 5)
                .map(i -> {
                    StockQuote q = new StockQuote();
                    q.setIndustryCode(industryCode);
                    q.setLanguageCode("en");
                    return q;
                })
                .delayElements(Duration.ofMillis(100));
    }

    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamData() {
        return Flux.interval(Duration.ofSeconds(1))
                .map(seq -> "Data item " + seq)
                .take(5);
    }
}

Essential difference — Blocking vs Non-Blocking:

// Approche BLOQUANTE (Spring MVC traditionnel)
@GetMapping("/hello")
public String getHello() {
    return "Hello World"; // Retourne directement
}

// Approche NON-BLOQUANTE (Spring WebFlux)
@GetMapping("/hello")
public Mono<String> getHello() {
    return Mono.just("Hello World"); // Retourne immédiatement un Publisher
    // Spring WebFlux souscrit et émet la réponse de manière asynchrone
}

• Mono.just(...) creates a Mono which emits a single value of type String.
• Unlike a traditional controller which returns the string directly, this method returns immediately with a Publisher.
• The actual string is sent asynchronously when Spring WebFlux subscribes to Publisher Mono, which then emits the event or element.

3.4 Internal architecture of Spring WebFlux

When an HTTP request arrives, here is what happens in detail:

HTTP Request
    │
    ▼
Reactor Netty (serveur HTTP non-bloquant)
    │  accepte la requête et la passe à Spring WebFlux
    ▼
DispatcherHandler (analogue au DispatcherServlet de Spring MVC)
    │
    ├── 1. Mappe la requête au bon Controller
    ├── 2. Invoque la méthode du Controller
    │        └─ Retourne Mono<T> ou Flux<T>
    ├── 3. Souscrit automatiquement au Mono/Flux retourné
    ├── 4. Collecte le résultat
    └── 5. Repasse au serveur (ex: Reactor Netty) pour écrire la réponse HTTP

Alternatives to Reactor Netty

3.5 Testing with WebTestClient

To test a WebFlux controller, we use WebTestClient — the reactive equivalent of MockMvc.

package com.example.demo;

import com.example.demo.controller.ReactionControllerOne;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.reactive.WebFluxTest;
import org.springframework.test.web.reactive.server.WebTestClient;

@WebFluxTest(ReactionControllerOne.class)
public class ReactionControllerOneTest {

    @Autowired
    WebTestClient webTestClient;

    @Test
    public void testMessage() {
        webTestClient.get().uri("/api/v1/message")
                .exchange()
                .expectStatus().isOk()
                .expectBody(String.class).isEqualTo("Sorry, couldn't help, but react!");
    }
}

Key points:

• @WebFluxTest(ReactionControllerOne.class) — Starts a Spring context limited to the specified controller (without starting a real server).
• @Autowired WebTestClient — Automatically injected, can bind to Controller and check responses without starting a server.
• webTestClient.get().uri(...) — Performs a non-blocking GET request.
• .exchange() — Execute the query.
• .expectStatus().isOk() — Checks that the HTTP status is 200.
• .expectBody(String.class).isEqualTo(...) — Checks the response body.

Because our Controller returns a Mono, the testing framework automatically subscribes and waits for the response behind the scenes. This basic pattern can be extended to more complex reactive pipelines.

Business value — Fintech case

By moving from a traditional blocking controller to the simple reactive approach with Mono<String>, you can:

• Increase the number of concurrent requests without increasing hardware support.
• Reduce response times during traffic peaks.
• Reduce cloud costs.
• Improve customer satisfaction.

Reactive programming translates directly into business value, especially in high-volume environments like finance, e-commerce, and real-time analytics.

3.6 Testing reactive streams with StepVerifier

StepVerifier is an essential tool for testing responsive streams in Java. Think of it as your personal QA inspector for responsive streams — it walks through the stream step by step, literally checking every signal and element along the way.

Main Methods of StepVerifier

Complete test example with StepVerifier

package com.example.demo;

import com.example.demo.operators.FlatMapOne;
import com.example.demo.operators.Industry;
import com.example.demo.operators.IndustryService;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

public class FlatMapOneTest {

    @Test
    public void testGetIndustryQuotes() {
        // Arrange : Utiliser le service stubé par défaut
        IndustryService stubService = new IndustryService();
        FlatMapOne flatMapOne = new FlatMapOne(stubService);

        // Act
        Flux<Industry> result = flatMapOne.getIndustryQuotes("software");

        // Assert
        StepVerifier.create(result)
                .expectNextMatches(industry ->
                        "software".equals(industry.getIndustryCode()) &&
                                industry.getStockQuoteList() != null &&
                                !industry.getStockQuoteList().isEmpty() &&
                                "software".equals(industry.getStockQuoteList().get(0).getIndustryCode())
                )
                .expectComplete()
                .verify();
    }
}

Test analysis:

1. We create an IndustryService stub and a FlatMapOne component.
2. We call getIndustryQuotes("software") — a method which returns a Stream<Industry>.
3. StepVerifier.create(result) — Subscribes to the Flow and starts observation.
4. expectNextMatches(...) — Checks that the first item emitted is an Industry with the correct code and a non-empty list of StockQuote.
5. expectComplete() — Checks that the Flow completes properly.
6. verify() — Effectively starts the verification.

3.7 Reactive operators — the pipeline

The reactive pipeline is the heart of what makes this style of programming so powerful. Each operator is a decorator that wraps the previous Flow by adding transformation or filtering behavior.

Design Pattern: This mechanism is the Decorator pattern (Gang of Four) in action. When you finally subscribe, the entire channel is activated.

Full Reactive Pipeline Demo

package com.example.demo.operators;

import org.jetbrains.annotations.NotNull;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.List;

public class ReactiveOperatorsExample {

    public static void main(String[] args) {
        new ReactiveOperatorsExample().demoReactivePipeline();
    }

    public void demoReactivePipeline() {

        // Simuler quelques secteurs industriels
        List<String> industries = List.of("software", "finance", "healthcare");

        Flux<StockQuote> industryFlux = Flux.fromIterable(industries)
                .delayElements(Duration.ofMillis(100)) // Simuler un délai async par élément
                .map(industryCode -> industryCode.toUpperCase()) // Transformer en majuscules
                .filter(code -> code.startsWith("S") || code.startsWith("F")) // Filtrer certains codes
                .flatMap(code -> getQuotesForIndustry(code)) // Aplatir les résultats
                .take(5)                        // Ne prendre que les 5 premières quotes
                .doOnNext(quote -> System.out.println("Processing quote: " + quote)) // Side effect
                .sort((q1, q2) -> q1.getIndustryCode().compareTo(q2.getIndustryCode())); // Tri

        // Exécuter (souscrire)
        industryFlux.blockLast(); // Uniquement pour la démo — bloquer pour voir l'output
    }

    private @NotNull Flux<StockQuote> getQuotesForIndustry(String industryCode) {
        StockQuote quote1 = new StockQuote();
        quote1.setIndustryCode(industryCode);
        quote1.setLanguageCode("en");

        StockQuote quote2 = new StockQuote();
        quote2.setIndustryCode(industryCode);
        quote2.setLanguageCode("fr");

        return Flux.just(quote1, quote2);
    }
}

Operator-by-operator analysis:

Important: These operators (map, filter, flatMap, etc.) do nothing immediately. They return a new Flow which wraps the previous one by adding behavior. The entire channel is only activated when you subscribe.

The flatMap operator with FlatMapOne

flatMap is particularly powerful: it takes each item in the stream, calls an asynchronous method that returns multiple StockQuotes for that industry, then flattens the results into the main stream.

package com.example.demo.operators;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;

@Component
public class FlatMapOne {

    private IndustryService service;

    @Autowired
    public FlatMapOne(IndustryService service) {
        this.service = service;
    }

    // Déclenché automatiquement au démarrage de l'application
    @EventListener(ContextRefreshedEvent.class)
    public void onApplicationEvent() {
        System.out.println(">> App is ready. Triggering getIndustryQuotes...");
        getIndustryQuotes("software")
                .doOnNext(System.out::println)
                .subscribe();
    }

    public Flux<Industry> getIndustryQuotes(String industryCode) {
        return service.getAllStockQuotes(industryCode, "en")
                .flatMap(e -> service.getAllStockQuotes(e.getIndustryCode(), "en")
                        .collectList()
                        .map(list -> new Industry(e.getIndustryCode(), list))
                );
    }
}

Pipeline explanation:

1. service.getAllStockQuotes(industryCode, "en") returns a Stream<StockQuote>.
2. For each StockQuote received, flatMap calls getAllStockQuotes again — retrieves the quotes for that industry code.
3. .collectList() collects all elements of the Stream into a List<StockQuote> (returns a Mono<List<StockQuote>>).
4. .map(list -> new Industry(...)) transforms this list into an Industry object.

3.8 Data Streaming with Server-Sent Events (SSE)

How does this look on the client side? Imagine a dashboard that displays fluctuating stock quotes in real time for a given industry. This could also be transaction lists, stock level events, or a continuous stream of weather data.

Controller with SSE

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamData() {
    return Flux.interval(Duration.ofSeconds(1))
            .map(seq -> "Data item " + seq)
            .take(5);
}

Explanation:

• Flux.interval(Duration.ofSeconds(1)) — Emits a new item every second.
• .map(seq -> "Data item " + seq) — Transform each Long into a String message.
• .take(5) — Emits exactly 5 items then completes the stream.
• MediaType.TEXT_EVENT_STREAM_VALUE — Tells Spring to stream this data as Server-Sent Events (SSE).

Test with curl

curl -N http://localhost:8080/api/v1/stream

Expected output (one item per second, not all at once):

data:Data item 0

data:Data item 1

data:Data item 2

data:Data item 3

data:Data item 4

Notice how each item arrives one by one, not all at once. It’s not a list — it’s a moving responsive stream.

What Spring WebFlux automatically handles with Flux

When using Flux, Spring WebFlux integrates backpressure aware Project Reactor publishers. It manages:

• dataflow control.
• Queue sizing.
• cancellation if client disconnects.

You don’t have to manage threads or buffers manually — that’s the magic of reactive streams.

Quick summary

3.9 Alternative — Virtual Threads

We have reached the grand finale. Let’s talk about the emerging alternative to reactive programming: virtual threads.

The problem that virtual threads solve

Virtual threads solve three main problems that reactive programming also addresses:

1. Prevent deadlock — which occurs when a thread calls a database, web service, file system, or network cache.
2. Reduce complexity — and allow a return to writing easy-to-read declarative code, compared to the tightly nested calls of a reactive pipeline.
3. Simplify debugging — it is difficult to track issues that arise in a complex reactive pipeline.

Platform Threads vs Virtual Threads

Virtual thread mechanism

Instead of using heavy platform threads (wrappers on OS threads, expensive to create, ~2 MB of RAM each), we rely on light virtual threads:

Virtual Thread (léger)
    │
    ├── Monté sur un Platform Thread quand il doit faire du travail CPU
    │
    └── Détecté par la JVM lors d'un blocage I/O
         │
         └── Swappé vers le heap (heap) — libère le Platform Thread
              │
              └── Platform Thread retourne au pool et gère d'autres requêtes

To reach 100% CPU utilization would potentially require a million platform threads, which would require ~2 TB of RAM and ~14 minutes to create them — this approach clearly doesn’t scale. Virtual threads solve this problem.

Implementation — BalanceController with Virtual Thread Executor

package com.example.demo.controller;

import com.example.demo.service.BalanceCheckService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

@RestController
public class BalanceController {

    private final BalanceCheckService balanceCheckService;
    // Exécuteur qui crée un virtual thread par tâche
    private final Executor virtualThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();

    public BalanceController(BalanceCheckService balanceCheckService) {
        this.balanceCheckService = balanceCheckService;
    }

    @GetMapping("/balance/{accountId}")
    public Mono<String> getBalance(@PathVariable String accountId) {
        return Mono.fromCallable(() -> {
            // S'exécute dans un Virtual Thread — pas besoin de chaînes réactives complexes
            return balanceCheckService.checkAccountBalance(accountId);
        }).subscribeOn(Schedulers.fromExecutor(virtualThreadExecutor));
    }
}

Key points:

• Executors.newVirtualThreadPerTaskExecutor() — Creates an executor that launches a new virtual thread for each submitted task.
• Mono.fromCallable(...) — Wraps a blocking call in a Mono.
• .subscribeOn(Schedulers.fromExecutor(virtualThreadExecutor)) — Tells Reactor to execute code on the virtual thread executor.

Simulated blocking service — BalanceCheckService

package com.example.demo.service;

import org.springframework.stereotype.Service;

@Service
public class BalanceCheckService {

    public String checkAccountBalance(String accountId) {
        try {
            // Simuler un appel bloquant (ex: requête JDBC, API externe)
            Thread.sleep(500); // 500 ms de délai
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "Error: Interrupted";
        }

        // Résultat simulé
        return "Account " + accountId + " balance: $1,234.56";
    }
}

Reactive Programming vs Virtual Threads — Comparison

Virtual threads move the complexity of reactive programming and its threading model from a framework like Spring WebFlux to the JDK, making the code much simpler to write and maintain.

Structured Concurrency (preview)

structured concurrency is the only functionality related to virtual threads still in preview status in the latest JDKs. It allows groups of threads to be treated as a single unit of work, further simplifying complex concurrent scenarios.

4. Référence du projet de démonstration

4.1 pom.xml — Maven Dependencies

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
                             https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.5.0</version>
        <relativePath/>
    </parent>

    <groupId>com.example</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>demo</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>21</java.version>
    </properties>

    <dependencies>
        <!-- Spring WebFlux — tire reactor-core automatiquement -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <!-- Tests Spring Boot -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!-- reactor-test : StepVerifier et WebTestClient -->
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!-- mockito-inline pour les mocks avancés -->
        <dependency>
            <groupId>org.mockito</groupId>
            <artifactId>mockito-inline</artifactId>
            <version>5.2.0</version>
            <scope>test</scope>
        </dependency>

        <!-- JetBrains annotations (@NotNull, etc.) -->
        <dependency>
            <groupId>org.jetbrains</groupId>
            <artifactId>annotations</artifactId>
            <version>RELEASE</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

Key dependencies:

4.2 Source structure

demo/
├── pom.xml
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/example/demo/
│   │   │       ├── DemoApplication.java              # Point d'entrée Spring Boot
│   │   │       ├── BackPressureOne.java              # Démonstration backpressure
│   │   │       ├── GnarlyReactiveOrderProcessor.java # Pipeline réactif avancé
│   │   │       ├── controller/
│   │   │       │   ├── ReactionControllerOne.java    # Controller WebFlux principal
│   │   │       │   └── BalanceController.java        # Controller avec virtual threads
│   │   │       ├── service/
│   │   │       │   └── BalanceCheckService.java      # Service bloquant simulé
│   │   │       └── operators/
│   │   │           ├── FlatMapOne.java               # Démonstration flatMap
│   │   │           ├── Industry.java                 # Modèle Industry
│   │   │           ├── IndustryService.java          # Service stub
│   │   │           ├── ReactiveOperatorsExample.java # Pipeline d'opérateurs
│   │   │           └── StockQuote.java               # Modèle StockQuote
│   │   └── resources/
│   │       └── application.properties               # Configuration Spring
│   └── test/
│       └── java/com/example/demo/
│           ├── DemoApplicationTests.java             # Test de contexte Spring
│           ├── FlatMapOneTest.java                   # Test StepVerifier
│           └── ReactionControllerOneTest.java        # Test WebTestClient
└── reactive_code_samples/
    └── src/
        └── Main.java                                 # Exemples de code réactif de base

4.3 Model classes

StockQuote

package com.example.demo.operators;

public class StockQuote {

    private String industryCode;
    private String languageCode;

    public String getIndustryCode() {
        return industryCode;
    }

    public void setIndustryCode(String industryCode) {
        this.industryCode = industryCode;
    }

    public String getLanguageCode() {
        return languageCode;
    }

    public void setLanguageCode(String languageCode) {
        this.languageCode = languageCode;
    }

    @Override
    public String toString() {
        return "StockQuote{" +
                "industryCode='" + industryCode + '\'' +
                ", languageCode='" + languageCode + '\'' +
                '}';
    }
}

Industry

package com.example.demo.operators;

import java.util.List;

public class Industry {

    private String industryCode;
    private List<StockQuote> stockQuoteList;

    public Industry(String industryCode, List<StockQuote> stockQuoteList) {
        this.industryCode = industryCode;
        this.stockQuoteList = stockQuoteList;
    }

    public String getIndustryCode() { return industryCode; }
    public void setIndustryCode(String industryCode) { this.industryCode = industryCode; }

    public List<StockQuote> getStockQuoteList() { return stockQuoteList; }
    public void setStockQuoteList(List<StockQuote> stockQuoteList) { this.stockQuoteList = stockQuoteList; }

    @Override
    public String toString() {
        return "Industry{" +
                "industryCode='" + industryCode + '\'' +
                ", stockQuoteList=" + stockQuoteList +
                '}';
    }
}

4.4 Services

IndustryService

package com.example.demo.operators;

import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

@Service
public class IndustryService {

    public Flux<StockQuote> getAllStockQuotes(String industryCode, String language) {
        StockQuote quote = new StockQuote();
        quote.setIndustryCode(industryCode);
        quote.setLanguageCode(language);
        return Flux.just(quote); // Données simulées
    }
}

application.properties

spring.application.name=demo

4.5 Components and listeners

DemoApplication — Spring Boot entry point

package com.example.demo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

4.6 Unit and integration testing

DemoApplicationTests — Context loading test

package com.example.demo;

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class DemoApplicationTests {

    @Test
    void contextLoads() {
        // Vérifie que le contexte Spring se charge sans erreur
    }
}

4.7 Advanced example — GnarlyReactiveOrderProcessor

This example illustrates a complex reactive command processing pipeline with retry, fallback, timeout, and error handling.

package com.example.demo;

import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.TimeoutException;

public class GnarlyReactiveOrderProcessor {

    public static void main(String[] args) {
        Order order = new Order(123L, true, true);
        new GnarlyReactiveOrderProcessor()
                .processOrder(order)
                .block();
    }

    public Mono<Order> processOrder(Order order) {
        return Mono.just(order)
                // Étape 1 : Validation avec retry à délai fixe
                .flatMap(this::validateOrder)
                .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(1)))

                // Étape 2 : Réservation d'inventaire avec backoff exponentiel et fallback
                .flatMap(validatedOrder -> {
                    if (validatedOrder.requiresInventoryReservation()) {
                        return reserveInventory(validatedOrder)
                                .retryWhen(Retry.backoff(5, Duration.ofMillis(500))
                                        .maxBackoff(Duration.ofSeconds(5)))
                                .onErrorResume(e -> fallbackToManualReservation(validatedOrder));
                    } else {
                        return Mono.just(validatedOrder);
                    }
                })

                // Étape 3 : Débit client avec backoff + jitter
                .flatMap(this::chargeCustomer)
                .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)).jitter(0.5))

                // Étape 4 : Expédition conditionnelle
                .flatMap(chargedOrder -> {
                    if (chargedOrder.shouldExpediteShipping()) {
                        return shipOrderExpedited(chargedOrder);
                    } else {
                        return shipOrderStandard(chargedOrder);
                    }
                })

                // Timeout global de 10 secondes avec notification de l'équipe ops
                .timeout(Duration.ofSeconds(10))
                .onErrorResume(TimeoutException.class,
                        e -> notifyOpsTeam(order, e).then(Mono.error(e)))

                // Étape 5 : Finalisation
                .flatMap(this::completeOrder)

                // Observabilité
                .doOnError(e -> System.err.println("Order processing failed for "
                        + order.getId() + ": " + e))
                .doOnSuccess(o -> System.out.println("Order " + o.getId()
                        + " processed successfully"));
    }

    // ... méthodes privées de validation, réservation, facturation, expédition
    private Mono<Order> validateOrder(Order order) {
        System.out.println("Validating order " + order.getId());
        return Mono.just(order).delayElement(Duration.ofMillis(200));
    }

    private Mono<Order> reserveInventory(Order order) {
        System.out.println("Reserving inventory for order " + order.getId());
        return Mono.just(order).delayElement(Duration.ofMillis(300));
    }

    private Mono<Order> fallbackToManualReservation(Order order) {
        System.out.println("Falling back to manual reservation for order " + order.getId());
        return Mono.just(order).delayElement(Duration.ofSeconds(1));
    }

    private Mono<Order> chargeCustomer(Order order) {
        System.out.println("Charging customer for order " + order.getId());
        return Mono.just(order).delayElement(Duration.ofMillis(250));
    }

    private Mono<Order> shipOrderExpedited(Order order) {
        System.out.println("Shipping order " + order.getId() + " with expedited shipping");
        return Mono.just(order).delayElement(Duration.ofMillis(400));
    }

    private Mono<Order> shipOrderStandard(Order order) {
        System.out.println("Shipping order " + order.getId() + " with standard shipping");
        return Mono.just(order).delayElement(Duration.ofMillis(400));
    }

    private Mono<Void> notifyOpsTeam(Order order, Throwable e) {
        System.out.println("Notifying ops team about timeout for order " + order.getId());
        return Mono.empty();
    }

    private Mono<Order> completeOrder(Order order) {
        System.out.println("Completing order " + order.getId());
        return Mono.just(order).delayElement(Duration.ofMillis(150));
    }

    // Classe Order interne
    static class Order {
        private final long id;
        private final boolean requiresInventoryReservation;
        private final boolean expediteShipping;

        public Order(long id, boolean requiresInventoryReservation, boolean expediteShipping) {
            this.id = id;
            this.requiresInventoryReservation = requiresInventoryReservation;
            this.expediteShipping = expediteShipping;
        }

        public long getId() { return id; }
        public boolean requiresInventoryReservation() { return requiresInventoryReservation; }
        public boolean shouldExpediteShipping() { return expediteShipping; }
    }
}

Advanced techniques demonstrated:

5. Versions and prerequisites

6. Summary of key concepts

Reactive Streams Interfaces

Types Project Reactor

Essential operators

Spring WebFlux annotations

Schedulers Reactor

Backpressure configuration

Search Terms

java · reactive · programming · backend · architecture · full-stack · web · spring · virtual · webflux · backpressure · threads · reactor · stepverifier · test · testing · boot · case · controllers · fintech · flux · mechanism · memory · operators