Building Non-Blocking Web Applications with Spring WebFlux and Project Reactor

A comprehensive guide to developing high-performance, event-driven web services using Spring WebFlux and the Project Reactor library for reactive programming in Java.

/ Article
Building Non-Blocking Web Applications with Spring WebFlux and Project Reactor
Photo by Mohammad Rahmani on Unsplash

Modern web applications often face demands for high concurrency and responsiveness. Traditional servlet-based architectures, while robust, can struggle with blocking I/O operations that tie up threads, limiting scalability. Reactive programming offers an alternative, enabling developers to build non-blocking, event-driven systems that handle many concurrent requests efficiently. This guide explores how to leverage Spring WebFlux and Project Reactor to construct such applications in Java.

Understanding Reactive Programming

Reactive programming is a paradigm focused on asynchronous data streams. It allows for the propagation of changes to interested parties. Think of it like a spreadsheet: when you change a cell, all dependent cells automatically update. In software, this means handling events and data flows over time, without blocking the execution thread.

The need for non-blocking I/O

In a traditional synchronous model, when an application performs an I/O operation (like reading from a database or calling an external API), the thread executing that operation waits until the I/O completes. During this wait, the thread cannot do any other work. If many concurrent requests involve such blocking operations, the server quickly runs out of available threads, leading to performance bottlenecks and reduced throughput.

Non-blocking I/O, conversely, allows a thread to initiate an I/O operation and then immediately return to other tasks. When the I/O operation finishes, it notifies the application, which can then process the result. This model enables a small number of threads to manage a large number of concurrent operations, significantly improving scalability and resource utilization.

Reactive Streams specification

The Reactive Streams specification provides a standard for asynchronous stream processing with non-blocking backpressure. It defines four core interfaces:

  • Publisher: Produces a sequence of items and sends them to Subscribers.
  • Subscriber: Consumes items produced by a Publisher.
  • Subscription: Represents the one-to-one relationship between a Publisher and a Subscriber. It is used by the Subscriber to request data or cancel the subscription.
  • Processor: Represents a processing stage that is both a Subscriber and a Publisher.

This specification ensures interoperability between different reactive libraries. Project Reactor is one such library that implements the Reactive Streams specification.

Introducing Project Reactor

Project Reactor is a foundational library for building reactive applications on the JVM. It provides two primary reactive types: Mono and Flux. These types represent asynchronous sequences of data.

Mono: zero or one element

A Mono<T> represents a stream that emits zero or one item, and then optionally completes with an error or a completion signal. It is suitable for operations that return a single result, like fetching a user by ID or saving a single entity.

Consider a simple Mono example:

import reactor.core.publisher.Mono;

public class MonoExample {
    public static void main(String[] args) {
        Mono<String> greetingMono = Mono.just("Hello, Reactor!");
        greetingMono.subscribe(
            data -> System.out.println("Received: " + data),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Completed!")
        );

        Mono<String> emptyMono = Mono.empty();
        emptyMono.subscribe(
            data -> System.out.println("Received: " + data),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Empty Mono Completed!")
        );
    }
}

In this example, greetingMono emits “Hello, Reactor!” and then completes. emptyMono completes immediately without emitting any data.

Flux: zero to N elements

A Flux<T> represents a stream that emits zero to N items, and then optionally completes with an error or a completion signal. It is used for operations that return multiple results, such as fetching a list of all users or a continuous stream of events.

Here is a Flux example:

import reactor.core.publisher.Flux;
import java.time.Duration;

public class FluxExample {
    public static void main(String[] args) throws InterruptedException {
        Flux<String> wordsFlux = Flux.just("Spring", "WebFlux", "Reactive", "Programming")
                                    .delayElements(Duration.ofMillis(100)); // Introduce a small delay

        wordsFlux.subscribe(
            word -> System.out.println("Emitted: " + word),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Flux Completed!")
        );

        // Keep the main thread alive to see the delayed emissions
        Thread.sleep(1000);
    }
}

This Flux emits four words sequentially, with a small delay between each. The subscribe method defines how to handle each emitted item, errors, and the completion signal.

Key operators for data transformation

Reactor provides a rich set of operators to transform, filter, and combine Mono and Flux streams. Some common operators include:

  • map: Transforms each item synchronously.
  • filter: Selectively includes items based on a predicate.
  • flatMap: Transforms each item into a new Publisher, then flattens these Publishers into a single sequence. This is crucial for asynchronous transformations.
  • zip: Combines the latest items from multiple Publishers into a single item.
  • concat: Concatenates multiple Publishers sequentially.
  • merge: Merges multiple Publishers into a single interleaved sequence.
Reactive Programming Flow
Photo by Mohammad Rahmani on Unsplash

Setting up a Spring Boot Project with WebFlux

Spring WebFlux is the reactive web framework included in Spring 5. It is built on Project Reactor and provides a non-blocking, event-driven alternative to Spring MVC.

Project dependencies

To start a new Spring Boot project with WebFlux, you can use Spring Initializr. Include the “Spring Reactive Web” dependency. If you are using Maven, your pom.xml will include:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
</dependency>

The spring-boot-starter-webflux transitively brings in Project Reactor and Netty, the default non-blocking server.

Creating a reactive REST endpoint

With WebFlux, you can define reactive endpoints using either annotation-based controllers (similar to Spring MVC) or functional endpoints. We will focus on annotation-based controllers for familiarity.

Consider a simple Product data class:

public class Product {
    private String id;
    private String name;
    private double price;

    // Constructors, getters, setters
    public Product(String id, String name, double price) {
        this.id = id;
        this.name = name;
        this.price = price;
    }

    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public double getPrice() { return price; }
    public void setPrice(double price) { this.price = price; }
}

Now, let’s create a reactive controller:

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
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;
import java.util.Arrays;
import java.util.List;

@RestController
@RequestMapping("/products")
public class ProductController {

    private final List<Product> products = Arrays.asList(
        new Product("1", "Laptop", 1200.00),
        new Product("2", "Mouse", 25.00),
        new Product("3", "Keyboard", 75.00),
        new Product("4", "Monitor", 300.00)
    );

    @GetMapping
    public Flux<Product> getAllProducts() {
        // Simulate a delay for each product to demonstrate reactive streaming
        return Flux.fromIterable(products)
                   .delayElements(Duration.ofMillis(500));
    }

    @GetMapping("/{id}")
    public Mono<Product> getProductById(@PathVariable String id) {
        return Mono.justOrEmpty(products.stream()
                                        .filter(p -> p.getId().equals(id))
                                        .findFirst())
                   .delayElement(Duration.ofSeconds(1)); // Simulate network latency
    }
}

In this controller:

  • getAllProducts() returns a Flux<Product>. When you access /products, the server will stream products one by one with a 500ms delay between each, rather than waiting for all to be ready.
  • getProductById() returns a Mono<Product>. It finds a product by ID and simulates a 1-second delay before emitting the single product or completing empty if not found.

Integrating with Reactive Data Sources

For a fully non-blocking application, the data layer must also be reactive. Spring Data provides reactive repositories for various databases.

Reactive Spring Data R2DBC

R2DBC (Reactive Relational Database Connectivity) is a specification for reactive programming with relational databases. Spring Data R2DBC provides implementations for several databases like PostgreSQL, H2, MySQL, and SQL Server.

To use R2DBC, add the appropriate starter dependency. For PostgreSQL, it would be:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-postgresql</artifactId>
</dependency>

Configure your application.properties for the R2DBC connection:

spring.r2dbc.url=r2dbc:postgresql://localhost:5432/mydatabase
spring.r2dbc.username=user
spring.r2dbc.password=password

Define a reactive repository interface:

import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public interface ProductRepository extends ReactiveCrudRepository<Product, String> {
    Flux<Product> findByNameContaining(String name);
    Mono<Product> findByPrice(double price);
}

Now, your ProductService (or directly your controller) can inject ProductRepository and use its reactive methods:

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

@Service
public class ProductService {

    private final ProductRepository productRepository;

    public ProductService(ProductRepository productRepository) {
        this.productRepository = productRepository;
    }

    public Flux<Product> findAllProducts() {
        return productRepository.findAll();
    }

    public Mono<Product> findProductById(String id) {
        return productRepository.findById(id);
    }

    public Mono<Product> saveProduct(Product product) {
        return productRepository.save(product);
    }
}

This setup ensures that database interactions are also non-blocking, maintaining the reactive chain throughout the application stack.

Error Handling in Reactive Streams

Errors can occur at any point in a reactive stream. Project Reactor provides operators to handle these errors gracefully.

  • onErrorResume: Allows you to provide a fallback Publisher when an error occurs. This can be useful for retrying operations or returning default data.
  • onErrorReturn: Returns a static fallback value when an error occurs.
  • doOnError: Executes a side-effect (like logging) when an error occurs, but does not consume or alter the error itself. The error continues downstream.
  • retry: Retries the upstream Publisher a specified number of times or indefinitely.

Example of error handling:

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

public class ErrorHandlingExample {
    public static void main(String[] args) {
        Flux.just("data1", "data2", "error", "data3")
            .flatMap(s -> {
                if ("error".equals(s)) {
                    return Mono.error(new RuntimeException("Simulated error!"));
                }
                return Mono.just(s.toUpperCase());
            })
            .onErrorResume(e -> {
                System.err.println("Resuming from error: " + e.getMessage());
                return Flux.just("fallback1", "fallback2"); // Provide fallback data
            })
            .subscribe(
                data -> System.out.println("Processed: " + data),
                error -> System.err.println("Final error: " + error.getMessage()),
                () -> System.out.println("Stream completed.")
            );

        System.out.println("\n--- Another example with onErrorReturn ---");

        Mono.just("value")
            .flatMap(s -> {
                if (Math.random() > 0.5) {
                    return Mono.error(new IllegalStateException("Random failure!"));
                }
                return Mono.just(s + " processed");
            })
            .onErrorReturn("default value on error") // Return a default value
            .subscribe(
                data -> System.out.println("Result: " + data),
                error -> System.err.println("This should not be called with onErrorReturn: " + error.getMessage()),
                () -> System.out.println("Mono completed.")
            );
    }
}

In the first example, when “error” is encountered, onErrorResume catches the RuntimeException and switches to emitting “fallback1” and “fallback2” before completing. In the second, onErrorReturn provides “default value on error” if the IllegalStateException occurs.

Testing Reactive Components with StepVerifier

Testing reactive streams requires a different approach than traditional synchronous code. Project Reactor provides StepVerifier, a utility for testing Mono and Flux sequences. It allows you to define an expectation for each event in the stream.

To use StepVerifier, ensure you have the reactor-test dependency in your pom.xml (it’s typically included with spring-boot-starter-webflux as a test scope dependency).

Example of testing a ProductService method:

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

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

public class ProductServiceTest {

    // Mock or simplified repository for testing
    private final List<Product> mockProducts = Arrays.asList(
        new Product("1", "Laptop", 1200.00),
        new Product("2", "Mouse", 25.00)
    );

    // Simulate a reactive repository
    private ProductRepository mockRepository = new ProductRepository() {
        @Override
        public <S extends Product> Mono<S> save(S entity) {
            return Mono.just(entity);
        }

        @Override
        public <S extends Product> Flux<S> saveAll(Iterable<S> entities) {
            return Flux.fromIterable(entities);
        }

        @Override
        public <S extends Product> Flux<S> saveAll(org.reactivestreams.Publisher<S> entityStream) {
            return Flux.from(entityStream);
        }

        @Override
        public Mono<Product> findById(String id) {
            return Mono.justOrEmpty(mockProducts.stream()
                                                .filter(p -> p.getId().equals(id))
                                                .findFirst());
        }

        @Override
        public Mono<Product> findById(org.reactivestreams.Publisher<String> id) {
            return Mono.from(id).flatMap(this::findById);
        }

        @Override
        public Flux<Product> findAll() {
            return Flux.fromIterable(mockProducts);
        }

        @Override
        public Flux<Product> findAllById(Iterable<String> ids) {
            return Flux.fromIterable(ids).flatMap(this::findById);
        }

        @Override
        public Flux<Product> findAllById(org.reactivestreams.Publisher<String> idStream) {
            return Flux.from(idStream).flatMap(this::findById);
        }

        @Override
        public Mono<Long> count() {
            return Mono.just((long) mockProducts.size());
        }

        @Override
        public Mono<Boolean> existsById(String id) {
            return findById(id).hasElement();
        }

        @Override
        public Mono<Boolean> existsById(org.reactivestreams.Publisher<String> id) {
            return Mono.from(id).flatMap(this::existsById);
        }

        @Override
        public Mono<Void> deleteById(String id) {
            return Mono.empty();
        }

        @Override
        public Mono<Void> deleteById(org.reactivestreams.Publisher<String> id) {
            return Mono.empty();
        }

        @Override
        public Mono<Void> delete(Product entity) {
            return Mono.empty();
        }

        @Override
        public Mono<Void> deleteAllById(Iterable<? extends String> ids) {
            return Mono.empty();
        }

        @Override
        public Mono<Void> deleteAll(Iterable<? extends Product> entities) {
            return Mono.empty();
        }

        @Override
        public Mono<Void> deleteAll(org.reactivestreams.Publisher<? extends Product> entityStream) {
            return Mono.empty();
        }

        @Override
        public Mono<Void> deleteAll() {
            return Mono.empty();
        }

        @Override
        public Flux<Product> findByNameContaining(String name) {
            return Flux.fromIterable(mockProducts)
                       .filter(p -> p.getName().contains(name));
        }

        @Override
        public Mono<Product> findByPrice(double price) {
            return Mono.justOrEmpty(mockProducts.stream()
                                                .filter(p -> p.getPrice() == price)
                                                .findFirst());
        }
    };

    private ProductService productService = new ProductService(mockRepository);

    @Test
    void testFindAllProducts() {
        StepVerifier.create(productService.findAllProducts())
            .expectNextCount(2) // Expect two products
            .verifyComplete(); // Verify the stream completes
    }

    @Test
    void testFindProductByIdFound() {
        StepVerifier.create(productService.findProductById("1"))
            .expectNextMatches(product -> product.getName().equals("Laptop"))
            .verifyComplete();
    }

    @Test
    void testFindProductByIdNotFound() {
        StepVerifier.create(productService.findProductById("99"))
            .expectNextCount(0) // Expect no elements
            .verifyComplete();
    }

    @Test
    void testSaveProduct() {
        Product newProduct = new Product("3", "Tablet", 500.00);
        StepVerifier.create(productService.saveProduct(newProduct))
            .expectNext(newProduct)
            .verifyComplete();
    }

    @Test
    void testFindByNameContaining() {
        StepVerifier.create(mockRepository.findByNameContaining("Mouse"))
            .expectNextMatches(product -> product.getName().equals("Mouse"))
            .verifyComplete();
    }
}

StepVerifier.create() takes the Publisher to be tested. You then chain methods like expectNext, expectNextCount, expectNextMatches, expectError, and verifyComplete to define the expected sequence of events. verifyComplete() triggers the subscription and verifies all expectations.

Practical Considerations and Best Practices

While reactive programming offers significant benefits, it also introduces a different way of thinking about application flow.

When to use reactive programming

Reactive programming with WebFlux is particularly well-suited for:

  • High-concurrency, I/O-bound applications: Microservices that frequently call other services or databases.
  • Streaming data: Applications that need to push data to clients in real-time or process continuous streams of data.
  • Event-driven architectures: Systems that react to a continuous flow of events.

It might be an over-engineering for simple CRUD applications with low concurrency requirements, where the overhead of reactive programming might outweigh the benefits.

Debugging challenges

Debugging reactive streams can be more complex than debugging synchronous code. The asynchronous nature and thread hopping make traditional stack traces less informative.

  • log() operator: Insert log() at various points in your stream to see the events flowing through it.
  • checkpoint() operator: Adds a description to the stack trace, making it easier to pinpoint where an error originated in a complex chain.
  • Hooks.onOperatorDebug(): Enables global operator debugging, providing more detailed stack traces for all operators. Use this sparingly in production due to performance impact.

Performance implications

Reactive applications can achieve higher throughput with fewer resources compared to traditional blocking applications, especially under heavy load. This is because threads are not blocked waiting for I/O. However, this does not necessarily mean individual request latency will be lower. The primary gain is in overall system capacity and efficiency.

Server Rack
Abigor, CC BY-SA 3.0 via Wikimedia Commons

Careful profiling and load testing are essential to confirm performance benefits for specific use cases. The non-blocking model shifts the bottleneck from thread contention to CPU processing or external service latency.

Conclusion

Spring WebFlux and Project Reactor provide a powerful foundation for building scalable, high-performance, non-blocking web applications. By embracing the reactive paradigm, developers can create systems that efficiently handle concurrent requests and stream data effectively. Understanding Mono and Flux, mastering operators, and integrating with reactive data sources are key steps in this journey. While it introduces a new mindset and some debugging complexities, the benefits in terms of scalability and resource utilization make reactive programming a valuable tool for modern application development.

Works Cited