Project Reactor is a Java library based on responsive stream specifications used to handle asynchronous data flows. Its core types are Mono and Flux. 1. Use operators such as map, flatMap, and filter for data flow conversion and processing. 2. Control data flow rate through backpressure mechanisms such as onBackpressureBuffer and onBackpressureDrop. 3. Use onErrorResume, onErrorReturn, and retry for error processing. 4. Use subscribeOn and publishOn to realize thread scheduling. 5. It is widely used in microservice asynchronous calls, event-driven architecture, real-time data processing and high concurrency API interfaces. Mastering responsive programming requires moving from imperative thinking to data flow-driven thinking, and ultimately implementing efficient and elegant asynchronous programming.

Reactive programming is becoming increasingly important in modern Java development, especially in handling asynchronous data flow scenarios with high concurrency and low latency. Project Reactor is a Java library based on Reactive Streams, developed and maintained by the Spring team, and is the underlying core of frameworks such as Spring WebFlux and Spring Cloud Gateway.

The following is a quick understanding of how to use Project Reactor for responsive programming in Java.

What is Project Reactor?

Project Reactor is a JVM library for building responsive applications. It implements the Reactive Streams specification, supports backpressure, and provides two core types:

• Mono<t></t> : represents an asynchronous sequence of 0 or 1 element (such as a single result or an empty result).
• Flux<t></t> : represents an asynchronous data stream (such as sets, event streams, etc.) of 0 to N elements.

They are all lazy : they are executed only when subscribed.

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

Flux.just("Apple", "Banana", "Cherry")
    .map(String::toUpperCase)
    .filter(s -> s.startsWith("B"))
    .subscribe(System.out::println);

Output:

 BANANA

Core concept: Operators

Reactor provides rich operators to handle data flows, common types include:

• Create classes : just , fromIterable , range , create
• Conversion classes : map , flatMap , concatMap
• Filter class : filter , take , skip
• Error handling : onErrorReturn , onErrorResume , retry
• Thread Scheduling : publishOn , subscribeOn

Example: flatMap for asynchronous flattening

 Flux.just("a", "b", "c")
    .flatMap(s -> Mono.just(s.toUpperCase())
        .delayElement(Duration.ofMillis(100))) // Simulate asynchronous call.subscribe(System.out::println);

In this example, each element is processed asynchronously, flatMap can handle multiple asynchronous tasks concurrently, which is suitable for calling HTTP services or databases.

How does Backpressure work?

In responsive streams, consumers control data flow rates to prevent producers from overwhelming consumers.

For example, you have a high-speed data source (such as a sensor) but the downstream processing is slower. Reactor will automatically "slow down" the upstream through the backpressure mechanism.

 Flux.range(1, 1000)
    .onBackpressureDrop() // If the downstream cannot process it, discard part of the data.publishOn(Schedulers.boundedElastic())
    .doOnNext(i -> {
        try { Thread.sleep(10); } catch (InterruptedException e) {}
        System.out.println("Processing: " i);
    })
    .blockLast();

Common backpressure strategies:

• onBackpressureBuffer : cache
• onBackpressureDrop : discard
• onBackpressureLatest : Only the latest one is kept

Error handling: Don't let exceptions interrupt the flow

Exceptions in responsive streams are not automatically thrown and must be handled explicitly.

 Mono.error(new RuntimeException("Boom!"))
    .onErrorResume(ex -> {
        System.err.println("Caught: " ex.getMessage());
        return Mono.just("Fallback Value");
    })
    .subscribe(System.out::println);

Output:

 Caught: Boom!
Fallback Value

Recommended strategies:

• onErrorResume : Returns an alternative value or alternate stream
• onErrorReturn : simply return the default value
• retry(2) : Retry up to 2 times

Thread Scheduling: subscribeOn vs publishOn

• subscribeOn : Decide where to start execution of the data stream
• publishOn : Decide which thread to execute the subsequent operation

 Flux.just("a", "b", "c")
    .map(s -> {
        System.out.println("Map 1: " Thread.currentThread().getName());
        return s.toUpperCase();
    })
    .publishOn(Schedulers.parallel()) // Switch thread.map(s -> {
        System.out.println("Map 2: " Thread.currentThread().getName());
        return s "!";
    })
    .subscribeOn(Schedulers.boundedElastic()) // Start thread.subscribe(System.out::println);

Thread.sleep(1000); // Prevent the main thread from quitting

The output will show that different stages are executed on different threads.

Practical application scenarios

1. Asynchronous call between microservices
   Use WebClient Mono/Flux to call other services non-blocking.

2. Event-driven architecture
   Handle event streams in message queues such as Kafka.

3. Real-time data processing
   Such as log flow, monitoring metrics, and user behavior flow.

4. High concurrency API interface
   Returns Mono<String> or Flux<User> in Spring WebFlux to improve throughput.

Tips: Debugging responsive streams

Responsive code debugging is difficult, recommended:

• Use .log() to view event streams:

   flux.log("MyFlux")

• Use .checkpoint() to locate the error stack:

   .flatMap(this::callService)
  .checkpoint("After calling service")

  ---

  Basically that's it. Reactor is not complicated, but it needs to change the "imperative" thinking to the "data flow-driven" approach. Once you master it, you will find it very powerful and elegant in handling asynchronous, streaming, high concurrency scenarios.

  The above is the detailed content of Reactive Programming in Java with Project Reactor. For more information, please follow other related articles on the PHP Chinese website!