EnglishEnglish العربيةالعربية PortuguêsPortuguês Japanese日本語
404 Not Found

404 Not Found

nginx

Spring WebFlux Reactive Programming

WebFlux is Spring's asynchronous engine—with a single Mono and first-class Flux, its non-blocking I/O handles tens of thousands of concurrent connections.

1. What You'll Learn

2. A True Story from a High-Concurrency Architect

(1) Pain Point: Thread Pool Exhaustion

Alice's OrderFlow encountered a bottleneck during a major sales event—with 500 concurrent users querying orders, the Tomcat thread pool (set to 200 threads by default) was exhausted, requests were queued, and the P99 latency skyrocketed to 3 seconds. Most of the time, threads were waiting for I/O (database, network), and CPU utilization was only 15%. Bob suggested increasing the number of threads, but adding 1,000 threads would consume an additional 1 GB of memory.

(2) The WebFlux Solution

WebFlux handles high concurrency with few threads—threads do not block during I/O waits and can continue serving other requests:

JAVA 
@GetMapping("/{id}")
public Mono<OrderResponse> getOrder(@PathVariable Long id) {
    return orderRepository.findById(id)
        .map(OrderResponse::from)
        .switchIfEmpty(Mono.error(new ResourceNotFoundException("Order", id)));
}

(3) Revenue

After Alice refactored the query API using WebFlux, four threads were able to handle 2,000 concurrent requests, memory usage was reduced by 70%, the P99 latency dropped from 3 seconds to 100 ms, and CPU utilization increased to 60%.

3. Core Concepts of Reactive Programming

(1) Mono and Flux

Type Meaning Analogy Example
Mono<T> 0 or 1 element Asynchronous version of Optional Query a single order
Flux<T> 0 to N elements Asynchronous version of Stream Query order list
Mono<Void> No return value Completion signal Delete operation
100% 
graph LR
    A["Publisher"] --> B["Mono<br/>0 or 1 element"]
    A --> C["Flux<br/>0 to N elements"]
    B --> D["onNext → onComplete"]
    C --> E["onNext x N → onComplete"]
    B --> F["onError"]
    C --> F

(1) ▶ Example: Basic Operations in Mono and Flux

JAVA 
// Mono: single value
Mono<String> mono = Mono.just("Hello OrderFlow");
Mono<String> empty = Mono.empty();
Mono<String> fromCallable = Mono.fromCallable(() -> fetchOrder(1L));

// Flux: multiple values
Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> range = Flux.range(1, 100);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

Output:

TEXT 
// Execution Successful

(2) Common Operators

Operator Function Example
map Conversion Elements .map(OrderResponse::from)
flatMap Asynchronous Conversion .flatMap(this::enrichOrder)
filter Filter .filter(order -> "PENDING".equals(order.getStatus()))
switchIfEmpty Null Substitution .switchIfEmpty(Mono.error(...))
onErrorResume Error Downgrade .onErrorResume(e -> fallbackOrder())
timeout Timeout Control .timeout(Duration.ofSeconds(5))
retry Retry .retry(3)
subscribeOn Specify Subscription Thread .subscribeOn(Schedulers.boundedElastic())

(2) ▶ Example: Operator Chaining

JAVA 
public Mono<OrderResponse> getOrderWithDetails(Long id) {
    return orderRepository.findById(id)
        .flatMap(order ->
            productRepository.findById(order.getProductId())
                .map(product -> OrderResponse.from(order, product)))
        .switchIfEmpty(Mono.error(
            new ResourceNotFoundException("Order", id)))
        .timeout(Duration.ofSeconds(3))
        .onErrorResume(TimeoutException.class,
            e -> Mono.error(new RuntimeException("Order query timed out")))
        .retry(2);
}

Output:

TEXT 
// Execution Successful

4. WebFlux REST API

(1) ▶ Example: Reactive Controller

JAVA 
@RestController
@RequestMapping("/api/v1/reactive/orders")
public class ReactiveOrderController {

    private final ReactiveOrderService orderService;

    public ReactiveOrderController(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @GetMapping("/{id}")
    public Mono<OrderResponse> getOrder(@PathVariable Long id) {
        return orderService.findById(id);
    }

    @GetMapping
    public Flux<OrderResponse> listOrders(
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {
        return orderService.findAll(page, size);
    }

    @PostMapping
    public Mono<ResponseEntity<OrderResponse>> createOrder(
            @Valid @RequestBody CreateOrderRequest request) {
        return orderService.create(request)
            .map(order -> ResponseEntity
                .status(HttpStatus.CREATED).body(order));
    }

    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> cancelOrder(@PathVariable Long id) {
        return orderService.cancel(id)
            .then(Mono.just(ResponseEntity.noContent().<Void>build()));
    }
}

Output:

TEXT 
// Execution Successful
Dimension WebMVC WebFlux
Return Type Synchronization Object Mono<T> / Flux<T>
Thread Model One thread per request Event loop (few threads)
I/O Model Blocking Non-blocking
Container Tomcat / Jetty Netty
Concurrency Limit Thread Pool Size Virtually Unlimited

5. Choosing Between WebFlux and WebMVC

(1) Product Selection Decision

Scenario Recommendation Reason
CRUD API, concurrency < 500 WebMVC Simple and intuitive, mature ecosystem
I/O-intensive, high concurrency WebFlux Non-blocking, supports high concurrency with few threads
Real-time streaming (SSE/WebSocket) WebFlux Native support
A large number of blocking operations (JDBC) WebMVC Blocking operations are actually worse in WebFlux
Mixed Scenarios WebMVC + Partial Asynchrony Progressive Optimization
⚠️ Note: WebFlux is not a panacea. If your code makes heavy use of blocking I/O (such as JDBC), WebFlux will actually perform worse than WebMVC. Choosing WebFlux means that the entire call chain must be non-blocking.

(1) ▶ Example: Server-Sent Events (SSE)

JAVA 
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<OrderEvent> streamOrderEvents() {
    return orderEventPublisher.eventStream()
        .log("order-events");
}

Output:

TEXT 
// Execution Successful

6. Spring Data R2DBC

(1) Responsive Database Access

(1) ▶ Example: R2DBC Entity and Repository

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

Output:

TEXT 
// Execution Successful
JAVA 
@Table("products")
public class Product {

    @Id
    private Long id;
    private String name;
    private BigDecimal price;
    private Integer stock;

    public Product() {}

    public Product(String name, BigDecimal price, Integer stock) {
        this.name = name;
        this.price = price;
        this.stock = stock;
    }
    // getters
}

public interface ReactiveProductRepository
        extends ReactiveCrudRepository<Product, Long> {

    Flux<Product> findByNameContaining(String keyword);

    Flux<Product> findByStockLessThan(Integer threshold);
}
Dimension JPA + JDBC R2DBC
Driver Model Blocking Non-blocking
API Style Synchronous Reactive (Mono/Flux)
Relationship Mapping Supported (@OneToMany) Not Supported
Query Language JPQL Native SQL / Method Name Derivation
Transaction @Transactional @Transactional (Reactive)

(2) ▶ Example: Responsive Service

JAVA 
@Service
public class ReactiveOrderService {

    private final ReactiveOrderRepository orderRepository;
    private final ReactiveProductRepository productRepository;

    public ReactiveOrderService(ReactiveOrderRepository orderRepository,
                                ReactiveProductRepository productRepository) {
        this.orderRepository = orderRepository;
        this.productRepository = productRepository;
    }

    public Mono<OrderResponse> create(CreateOrderRequest request) {
        return productRepository.findById(request.productId())
            .switchIfEmpty(Mono.error(
                new ResourceNotFoundException("Product", request.productId())))
            .flatMap(product -> {
                if (product.getStock() < request.quantity()) {
                    return Mono.error(new InsufficientStockException(
                        product.getId(), product.getStock(), request.quantity()));
                }
                product.deductStock(request.quantity());
                return productRepository.save(product)
                    .then(orderRepository.save(new Order(product.getId(), request.quantity())));
            })
            .map(OrderResponse::from);
    }

    public Flux<OrderResponse> findAll(int page, int size) {
        return orderRepository.findAll()
            .skip(page * size)
            .take(size)
            .map(OrderResponse::from);
    }

    public Mono<Void> cancel(Long orderId) {
        return orderRepository.findById(orderId)
            .flatMap(order -> {
                order.setStatus("CANCELLED");
                return orderRepository.save(order);
            })
            .then();
    }
}

Output:

TEXT 
// Execution Successful

7. Comprehensive Example: OrderFlow Reactive Query API

JAVA 
// ReactiveProductRepository.java
public interface ReactiveProductRepository
        extends ReactiveCrudRepository<Product, Long> {
    Flux<Product> findByNameContaining(String keyword);
}

// ReactiveOrderRepository.java
public interface ReactiveOrderRepository
        extends ReactiveCrudRepository<Order, Long> {
    Flux<Order> findByStatus(String status);
}

// ReactiveOrderController.java
@RestController
@RequestMapping("/api/v1/reactive/orders")
public class ReactiveOrderController {

    private final ReactiveOrderService orderService;

    public ReactiveOrderController(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @GetMapping("/{id}")
    public Mono<OrderResponse> get(@PathVariable Long id) {
        return orderService.findById(id);
    }

    @GetMapping
    public Flux<OrderResponse> list(
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {
        return orderService.findAll(page, size);
    }

    @GetMapping("/status/{status}")
    public Flux<OrderResponse> byStatus(@PathVariable String status) {
        return orderService.findByStatus(status);
    }

    @PostMapping
    public Mono<ResponseEntity<OrderResponse>> create(
            @RequestBody CreateOrderRequest req) {
        return orderService.create(req)
            .map(o -> ResponseEntity.status(HttpStatus.CREATED).body(o));
    }

    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> cancel(@PathVariable Long id) {
        return orderService.cancel(id)
            .thenReturn(ResponseEntity.noContent().<Void>build());
    }
}

// ReactiveOrderService.java
@Service
public class ReactiveOrderService {

    private final ReactiveOrderRepository orderRepo;
    private final ReactiveProductRepository productRepo;

    public ReactiveOrderService(ReactiveOrderRepository orderRepo,
                                ReactiveProductRepository productRepo) {
        this.orderRepo = orderRepo;
        this.productRepo = productRepo;
    }

    public Mono<OrderResponse> findById(Long id) {
        return orderRepo.findById(id)
            .map(OrderResponse::from)
            .switchIfEmpty(Mono.error(new ResourceNotFoundException("Order", id)));
    }

    public Flux<OrderResponse> findAll(int page, int size) {
        return orderRepo.findAll().skip((long) page * size).take(size)
            .map(OrderResponse::from);
    }

    public Flux<OrderResponse> findByStatus(String status) {
        return orderRepo.findByStatus(status).map(OrderResponse::from);
    }

    public Mono<OrderResponse> create(CreateOrderRequest req) {
        return productRepo.findById(req.productId())
            .switchIfEmpty(Mono.error(new ResourceNotFoundException("Product", req.productId())))
            .flatMap(p -> {
                if (p.getStock() < req.quantity()) {
                    return Mono.error(new InsufficientStockException(p.getId(), p.getStock(), req.quantity()));
                }
                p.deductStock(req.quantity());
                return productRepo.save(p)
                    .then(orderRepo.save(new Order(p.getId(), req.quantity())));
            })
            .map(OrderResponse::from);
    }

    public Mono<Void> cancel(Long id) {
        return orderRepo.findById(id)
            .flatMap(o -> { o.setStatus("CANCELLED"); return orderRepo.save(o); })
            .then();
    }
}

❓ FAQ

Q Can WebFlux completely replace WebMVC?
A We do not recommend it. The WebMVC ecosystem is more mature, and for the vast majority of CRUD projects, WebMVC is simpler to use. WebFlux is suitable for I/O-intensive, high-concurrency, and real-time streaming scenarios. The two can coexist, but they should not be mixed (do not use blocking code within WebFlux).
Q What is the difference between Mono and CompletableFuture?
A Mono is lazy (it executes only when subscribed to), while CompletableFuture is eager (it starts executing as soon as it is created). Mono supports backpressure and cancellation, whereas CompletableFuture does not.
Q Does R2DBC support @OneToMany relationship mapping?
A No, it does not. R2DBC is a lightweight reactive database driver that does not perform relationship mapping. When you need to perform join queries, you must manually combine multiple queries using flatMap.
Q How do I call blocking code in WebFlux?
A Use Mono.fromCallable() + subscribeOn(Schedulers.boundedElastic()) to schedule blocking operations to the elastic thread pool. However, frequent use indicates that this approach is not suitable for WebFlux.
Q How is error handling done in WebFlux?
A 1) Operators onErrorResume / onErrorReturn; 2) Global @ExceptionHandler (also supported by WebFlux); 3) onErrorMap to convert exception types.
Q How do you test WebFlux?
A Use WebTestClient instead of MockMvc. @WebFluxTest loads the WebFlux layer. StepVerifier is used to test element sequences in Mono/Flux.

📖 Summary

📝 Exercises

  1. Basic Problem (Difficulty ⭐): Implement a reactive API for product queries using WebFlux (GET /api/v1/reactive/products/{id}), connecting to an H2 in-memory database via R2DBC.

  2. Advanced Exercise (Difficulty: ⭐⭐): Implement a reactive order placement API that covers the entire process—including inventory checking, inventory deduction, and order creation—using flatMap to combine multiple reactive operations.

  3. Challenge (Difficulty: ⭐⭐⭐): Use WebTestClient and StepVerifier to write a reactive API test that implements an SSE real-time order event push interface, and compare the performance differences between WebMVC and WebFlux under 1,000 concurrent connections.

Web-Tutorial.com

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站，几个人的小团队，精力有限。页面虽经检查，难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了，麻烦告诉我们，我们会在收到反馈后第一时间进行修复，再次感谢您的光临 🙏