In the previous tutorial, I introduced to you all about Reactive Streams in Java which implements Reactive Streams specification. In this tutorial, I will introduce you all one more library that also implements the Reactive Streams specification: it is Project Reactor. How is it in detail? Just continue to read this tutorial.
First, as with any tutorial, I will also create a Maven project as an example:
To work with Project Reactor, we need to add its dependency:
|
1 2 3 4 5 |
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> <version>3.5.9</version> </dependency> |
Because of the Reactive Streams specification, we also have Publisher, Subscriber, and Subscription concepts in Project Reactor, but with different names.
In Project Reactor, we have two objects that act as Publisher:
- One is the Flux object: This object can publish from 0 to N items.
- Second is the Mono object: This object can only publish 1 item or no item.
The difference between these two objects is: if your code returns a Mono object then we can know that this object can only publish up to one item, and for a Flux object we have to code to our application that can handle up to N items. From there, we have a way to develop our application properly.
For Subscriber, we can use an abstract class called BaseSubscriber that implements Subscriber. This class defines predefined methods, like hookOnSubscribe(), hookOnNext(), hookOnComplete(), hookOnError(), hookOnCancel(), and hookFinally(), to help us can develop the Subscriber part of the Reactive Streams application.
OK, now let’s try an example.
First, we will define an object to extend BaseSubscriber abstract class to implement the Subscriber part.
Its object is as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
package com.huongdanjava.projectreactor; import org.reactivestreams.Subscription; import reactor.core.publisher.BaseSubscriber; public class Consumer extends BaseSubscriber<String> { @Override protected void hookOnSubscribe(Subscription subscription) { System.out.println("Consumer: hookOnSubscribe."); request(1); } @Override protected void hookOnNext(String value) { System.out.println("Consumer: hookOnNext " + value); request(1); } } |
In this Consumer object, as soon as subscribing to Publisher, it will request to retrieve the first item. And every time the Publisher publishes new data, it also requests to retrieve that data.
Next, I will define a Publisher using the Flux object to publish the data.
Flux has many static methods for creating data such as just(), fromArray(), fromIterable(), … In this tutorial, I just use the simple as follows:
|
1 |
Flux<String> flux = Flux.just("Khanh", "Quan", "Thanh"); |
The last step is: subscribe our Consumer object to Publisher.
|
1 |
flux.subscribe(new Consumer()); |
The entire code is as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
package com.huongdanjava.projectreactor; import java.util.concurrent.TimeUnit; import reactor.core.publisher.Flux; public class Example { public static void main(String[] args) throws InterruptedException { Flux<String> flux = Flux.just("Khanh", "Quan", "Thanh"); flux.subscribe(new Consumer()); TimeUnit.SECONDS.sleep(10); } } |
Result:
