Usually, we write Reactive applications that will run on the main thread but there are cases where we need to use multithread. For example, after receiving an item from the Publisher, the Consumer needs to process something different from that data item. That manipulation, we can run on another thread to increase performance. In Project Reactor, to use multithread we can use the Scheduler object. How is it in details? Let’s find out in this tutorial.
First, I will create a new Maven project as an example:
with Project Reactor dependency as follows:
|
1 2 3 4 5 |
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> <version>3.1.8.RELEASE</version> </dependency> |
In the example of this tutorial, I am going to create a small application with Publisher that can generate numbers, then we will use multithread to duplicate the values of these numbers, eventually going back to the user the results after processing.
But first let’s see what our code will look like in case we do not use multithread. The code will look like this:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
package com.huongdanjava.projectreactor; import reactor.core.publisher.Flux; public class Application { public static void main(String[] args) { Flux.just(5, 10, 23, 25).log() .flatMap(n -> Flux.just(n * 2)) .subscribe(System.out::println); } } |
Here, I used the Flux object to emit a number and used the flatMap() method to convert that value to double and use the other Flux object to output the value after the conversion to the user. The Consumer will subscribe to the final value after conversion for printing to the console.
The result of running the above example is as follows:
As you can see, our code is currently running on the main thread only.
Now we will use the Scheduler object to run multithread.
There are two methods help we can run Reactive applications with Project Reactor using multithread: subscribeOn() and publishOn(). The difference between these two methods is that the subscribeOn() method determines whether the subscription process will take place in which thread (main thread or new thread), and the publishOn() method will determine the consumer in similar meaning.
Parameters of these two methods are the Scheduler object as I said above. This object will return the thread that we need. We can use Schedulers class to initialize the Scheduler object, with some strategies as follows:
- Schedulers.parallel(): Use a fixed threadpool to perform parallel tasks.
- Schedulers.elastic(): The thread pool mechanism can be expanded if the current threads is used up.
- Schedulers.from(Executor): use custom Executor of Java.
- Schedulers.single(): Same as Schedulers.parallel () but here only 1 thread.
OK, let’s say I now add the subscribeOn() method to our example above:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
package com.huongdanjava.projectreactor; import reactor.core.publisher.Flux; import reactor.core.scheduler.Schedulers; public class Application { public static void main(String[] args) throws InterruptedException { Flux.just(5, 10, 23, 25).log() .flatMap(n -> Flux.just(n * 2)).subscribeOn(Schedulers.elastic()) .subscribe(value -> System.out.println(Thread.currentThread().getName()+ " : " + value)); Thread.sleep(10000); } } |
when running, the result will look like this:
Obviously, as you can see, our code no longer runs on the main thread anymore.
If you now modify the subscribeOn() method to publishOn():
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
package com.huongdanjava.projectreactor; import reactor.core.publisher.Flux; import reactor.core.scheduler.Schedulers; public class Application { public static void main(String[] args) throws InterruptedException { Flux.just(5, 10, 23, 25).log() .flatMap(n -> Flux.just(n * 2)).publishOn(Schedulers.elastic()) .subscribe(value -> System.out.println(Thread.currentThread().getName()+ " : " + value)); Thread.sleep(10000); } } |
You will see the results as follows:
As you can see, our consumer now runs at the “elastic-2” thread, not the main thread anymore.
