In this blog, we’ll get introduced to the Spring Reactor project and its importance. The idea is to take advantage of the Reactive Streams Specification to build non-blocking reactive applications on the JVM.
Using this knowledge, we’ll build a simple reactive application and compare it to a traditional blocking application.
Reactive applications are the “hot new thing” making a lot of applications switch to this model. You can read more about this in The Reactive Manifesto.
Conventional APIs are Blocking
Modern applications deal with a high number of concurrent users and data. Moore’s law is no longer holding as it used to. The hardware capabilities, although increasing, are not keeping up with modern applications where performance is very important.
Java developers by default write blocking code. It’s just how the API was set up. Another example would be the traditional servlet (Tomcat) approach. Each request warrants a new thread which waits for the whole background process to finish in order to send the response back.
This means that our data layer logic is blocking the application by default since Threads idly wait for a response. It’s wasteful to not reuse these Threads for a different purpose, while we wait for the response to get back.
Note: This might be a problem if we have limited resources or a process takes too much time to execute.
Asynchronous still Blocks
In Java, you can write code asynchronously using Callbacks and Futures. You can then get and join threads at some later point in time and process the result. Java 8 introduced us with a new class – CompletableFuture, which makes it much easier to coordinate these things.
It works in a simple fashion – when a single process ends, another one starts. After the second one ends, the results are combined into a third process.
This makes it a lot easier to coordinate your application, but it’s still ultimately blocking as it creates Threads and waits upon calling a .join() method.
Reactive Programming
What we want is asynchronous and non-blocking. A group of developers from companies like Netflix, Pivotal, RedHat, etc. got together and converged on something called The Reactive Streams Specification.
Project Reactor is Spring’s implementation of The Reactive Specification and it’s specifically favored by the Spring Webflux module, although you can use it with other modules like RxJava.
The idea is to operate Asynchronously with Backpressure using Publishers and Subscribers.
Here, we’re being introduced to several new concepts! Let’s explain them one by one:
- Publisher – A Publisher is a provider of a potentially unbounded number of elements.
- Subscriber – A Subscriber listens to that Publisher, asking for new data. Sometimes, it’s also referred to as a Consumer.
- Backpressure – The ability of the Subscriber to let the Publisher how many requests can it handle at the time. So it’s the Subscriber that is responsible for the flow of the data, not the Publisher as it just provides the data.
The Reactor Project offers 2 types of publishers. These are considered the main building blocks of Spring Webflux:
- Flux – is a publisher that produces
0toNvalues. It could be unbounded. Operations that return multiple elements use this type. - Mono – is a publisher that produces
0to1value. Operations that return a single element use this type.
Developing Reactive Applications
With all of the above in mind, let’s jump into creating a simple web application and take advantage over this new reactive paradigm!
The simplest way to start with a skeleton Spring Boot project, as always, is using Spring Initializr. Select your preferred version of Spring Boot and add the “Reactive Web” dependency. After this, generate it as a Maven project and you’re all set!
Let’s define a simple POJO – Greeting:
public class Greeting {private String msg;// Constructors, getters and setters}
Defining a Publisher
Alongside it, let’s define a simple REST Controller with an adequate mapping:
@RestControllerpublic class GreetReactiveController {@GetMapping("/greetings")public Publisher<Greeting> greetingPublisher() {Flux<Greeting> greetingFlux = Flux.<Greeting>generate(sink -> sink.next(new Greeting("Hello"))).take(50);return greetingFlux;}}
Calling Flux.generate() will create a never ending stream of the Greeting object.
The take() method, as the name suggests, will only take first 50 values from the stream.
It’s important to note that the return type of the method is the asynchronous type Publisher<Greeting>.
To test this endpoint, navigate your browser to http://localhost:8080/greetings or use the curl client on your command line – curl localhost:8080/greetings
Server-Sent Events
Another publisher that has been used ever since their arrival is Server-Sent Events.
These events allow a web page to get updates from a server in real-time.
Let’s define a simple reactive server:
@GetMapping(value = "/greetings/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Publisher<Greeting> sseGreetings() {Flux<Greeting> delayElements = Flux.<Greeting>generate(sink -> sink.next(new Greeting("Hello @" + Instant.now().toString()))).delayElements(Duration.ofSeconds(1));return delayElements;}
Alternatively, we could’ve defined this:
@GetMapping(value = "/greetings/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)Flux<Greeting> events() {Flux<Greeting> greetingFlux = Flux.fromStream(Stream.generate(() -> new Greeting("Hello @" + Instant.now().toString())));Flux<Long> durationFlux = Flux.interval(Duration.ofSeconds(1));return Flux.zip(greetingFlux, durationFlux).map(Tuple2::getT1);}
Defining a Consumer
Now let’s see the consumer side of it. It’s worth noting that you don’t need to have a reactive publisher in order to use reactive programming on the consuming side:
public class Person {private int id;private String name;// Constructor with getters and setters}
And then we have a traditional RestController with a single mapping:
@RestControllerpublic class PersonController {private static List<Person> personList = new ArrayList<>();static {personList.add(new Person(1, "John"));personList.add(new Person(2, "Jane"));personList.add(new Person(3, "Max"));personList.add(new Person(4, "Alex"));personList.add(new Person(5, "Aloy"));personList.add(new Person(6, "Sarah"));}@GetMapping("/person/{id}")public Person getPerson(@PathVariable int id, @RequestParam(defaultValue = "2") int delay)throws InterruptedException {Thread.sleep(delay * 1000);return personList.stream().filter((person) -> person.getId() == id).findFirst().get();}}
Our producer app is running on port 8080. Now let’s say that we want to call the /person/{id} endpoint 5 times. We know that, by default, each response takes a 2-second delay due to “network lag”.
Using RestTemplate
As expected it took a little over 10 secs and this is how Spring MVC works by default.
At this day and age, waiting for a little over 10 seconds for a result on a page is unacceptable. This is the difference between keeping a customer/client and losing it due to waiting for too long.
Spring Reactor introduced a new web client to make web requests called WebClient. Compared to RestTemplate, this client has a more functional feel and is fully reactive. It’s included in the spring-boot-starter-weblux dependency and it’s build to replace RestTemplate in a non-blocking way.
Let’s rewrite the same controller, this time, using WebClient:
And nothing happened, as expected.
This is because we are not subscribing . It’s asynchronous but it also doesn’t kick off until we call the .subscribe() method. This is a common problem with people who are new to Spring Reactor, so keep an eye out for this.
Let’s change our main method and add subscribe:
for (int i = 1; i <= 5; i++) {client.get().uri("/person/{id}", i).retrieve().bodyToMono(Person.class).subscribe();}
Adding the method prompts us with the wanted result:
The request is sent but the .subscribe() method doesn’t sit and wait for the response. Since it doesn’t block, it finished before receiving the response at all.
Could we counter this by chaining .block() at the end of the method calls?
for (int i = 1; i <= 5; i++) {client.get().uri("/person/{id}", i).retrieve().bodyToMono(Person.class).block();}
The way to fix all of these problems is simple: We make a list of type Mono and wait for all of them to complete, rather than waiting for each one:
List<Mono<Person>> list = Stream.of(1, 2, 3, 4, 5).map(i -> client.get().uri("/person/{id}", i).retrieve().bodyToMono(Person.class)).collect(Collectors.toList());Mono.when(list).block();
This is what we’re aiming for. This time, it took just over two seconds, even with massive network lag. This increases the efficiency of our application drastically and really is a game-changer.
If you look closely at the threads, Reactor is reusing them rather than creating new ones. This is really important if your application handles many requests in a short span of time.
Conclusion
In this blog, we discussed the need for reactive programming and Spring’s implementation of it – the Spring Reactor.
References :-
