{{announcement.body}}
{{announcement.title}}

Build Reactive REST APIs With Spring WebFlux

DZone 's Guide to

Build Reactive REST APIs With Spring WebFlux

See how to Build Reactive REST APIs with Spring WebFlux. This explains the advantages we get with Reactive APIs and the reactive specification.

· Java Zone ·
Free Resource

In this article, we will see how to build reactive REST APIs with Spring WebFlux. Before jumping into the reactive APIs, let us see how the systems evolved, what problems we see with the traditional REST implementations, and the demands from modern APIs.

If you look at the expectations from legacy systems to modern systems described below,

Evolution of application API needs

The expectations from the modern systems are: the applications should be distributed, Cloud Native, embracing for high availability, and scalability. So the efficient usage of system resources is essential. Before jumping into Why reactive programming to build REST APIs? Let us see how the traditional REST APIs request processing works.

Traditional REST API model

Below are the issues what we have with the traditional REST APIs,

  • Blocking and Synchronous → The request is blocking and synchronous. The request thread will be waiting for any blocking I/O and the thread is not freed to return the response to the caller until the I/O wait is over.
  • Thread per request → The web container uses a thread-per-request model. This limits the number of concurrent requests to handle. Beyond certain requests, the container queues the requests that eventually impact the performance of the APIs.
  • Limitations to handle high concurrent users → As the web container uses a thread-per-request model, we cannot handle high concurrent requests.
  • No better utilization of system resources → The threads will be blocking for I/O and sitting idle. But, the web container cannot accept more requests. During this scenario, we are not able to utilize the system resources efficiently.
  • No backpressure support → We cannot apply backpressure from the client or the server. If there is a sudden surge of requests the server or client outages may happen. After that, the application will not be accessible to the users. If we have backpressure support, the application should sustain during the heavy load rather than the unavailability.

Let us see how we can solve the above issues using reactive programming. Below are the advantages we will get with reactive APIs.

  • Asynchronous and Non-Blocking → Reactive programming gives the flexibility to write asynchronous and Non-Blocking applications.
  • Event/Message Driven→ The system will generate events or messages for any activity. For example, the data coming from the database is treated as a stream of events.
  • Support for backpressure → Gracefully we can handle the pressure from one system to on to the other system by applying back pressure to avoid denial of service.
  • Predictable application response time → As the threads are asynchronous and non-blocking, the application response time is predictable under the load.
  • Better utilization of system resources → As the threads are asynchronous and non-blocking, the threads will not be hogged for the I/O. With fewer threads, we could able to support more user requests.
  • Scale based on the load
  • Move away from thread per request → With the reactive APIs, we are moving away from thread per request model as the threads are asynchronous and non-blocking. Once the request is made, it creates an event with the server and the request thread will be released to handle other requests.

Now, let us see how Reactive Programming works. In the below example, once the application makes a call to get the data from a data source, the thread will be returned immediately, and the data from the data source will come as a data/event stream. Here, the application is a subscriber, and the data source is a publisher. Upon the completion of the data stream, the onComplete event will be triggered.

Data stream workflow

Below is another scenario where the publisher will trigger an onError event if any exception happens.
Data stream workflow

In some cases, there might not be any items to deliver from the publisher. For example, deleting an item from the database. In that case, the publisher will trigger the onComplete/onError event immediately without calling onNext event, as there is no data to return.

Data stream workflow

Now, let us see what is backpressure and how we can apply backpressure to the reactive streams. For example, we have a client application that is requesting data from another service. The service is able to publish the events at the rate of 1000TPS but the client application is able to process the events at the rate of 200TPS. 

In this case, the client application should buffer the rest of the data to process. Over the subsequent calls, the client application may buffer more data and eventually run out of memory. This causes the cascading effect on the other applications which depends on the client application. To avoid this the client application can ask the service to buffer the events at their end and push the events at the rate of the client application. This is called backpressure. The below diagram depicts the same. 


Now, We will see the reactive streams specification and one of its implementation called Project Reactor. Reactive Streams specification has the following interfaces defined. Let us see the details of those interfaces.

Publisher → A Publisher is a provider of a potentially unlimited number of sequenced elements, publishing them as requested by its Subscriber(s)

Java
 




x


 
1
public interface Publisher<T> {
2
    public void subscribe(Subscriber<? super T> s);
3
}



Subscriber → A Subscriber is a consumer of a potentially unbounded number of sequenced elements.

Java
 




xxxxxxxxxx
1


 
1
public interface Subscriber<T> {
2
    public void onSubscribe(Subscription s);
3
    public void onNext(T t);
4
    public void onError(Throwable t);
5
    public void onComplete();
6
}



Subscription → A Subscription represents a one-to-one lifecycle of a Subscriber subscribing to a Publisher.

Java
 




xxxxxxxxxx
1


 
1
public interface Subscription {
2
    public void request(long n);
3
    public void cancel();
4
}



Processor → A Processor represents a processing stage — which is both a Subscriber and a Publisher and obeys the contracts of both.

The class diagram of the reactive streams specification is given below.
Reactive stream specification

The reactive streams specification has many implementations. Project Reactor is one of the implementations. The Reactor is fully non-blocking and provides efficient demand management. The Reactor offers two reactive and composable APIs, Flux [N], and Mono [0|1], which extensively implement Reactive Extensions. Reactor offers Non-Blocking, backpressure-ready network engines for HTTP (including Websockets), TCP, and UDP. It is well-suited for a microservices architecture.

Flux → It is a Reactive Streams Publisher with rx operators that emits 0 to N elements, and then complete (successfully or with an error). The marble diagram of the Flux is represented below.

  • Mono→ It is a Reactive Streams Publisher with basic rx operators that completes successfully by emitting 0 to 1 element, or with an error. The marble diagram of the Mono is represented below.

Image Credit: https://projectreactor.io

As Spring 5.x comes with Reactor implementation, if we want to build REST APIs using imperative style programming with Spring servlet stack, it still supports. Below is the diagram which explains how Spring supports both reactive and servlet stack implementations.



Image Credit: spring.io

Now, we will see an application to expose reactive REST APIs. In this application, we used:

  • Spring Boot with WebFlux
  • Spring Data for Cassandra with Reactive Support
  • Cassandra Database

Below is the high-level architecture of the application.
Reactive demo application workflow

Let us look at the build.gradle file to see what dependencies are included to work with the Spring WebFlux.

Groovy
 




xxxxxxxxxx
1
26


 
1
plugins {
2
    id 'org.springframework.boot' version '2.2.6.RELEASE'
3
    id 'io.spring.dependency-management' version '1.0.9.RELEASE'
4
    id 'java'
5
}
6
 
           
7
group = 'org.smarttechie'
8
version = '0.0.1-SNAPSHOT'
9
sourceCompatibility = '1.8'
10
 
           
11
repositories {
12
    mavenCentral()
13
}
14
 
           
15
dependencies {
16
    implementation 'org.springframework.boot:spring-boot-starter-data-cassandra-reactive'
17
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
18
    testImplementation('org.springframework.boot:spring-boot-starter-test') {
19
        exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
20
    }
21
    testImplementation 'io.projectreactor:reactor-test'
22
}
23
 
           
24
test {
25
    useJUnitPlatform()
26
}



In this application, I have exposed the below mentioned APIs. You can download the source code from GitHub.
Endpoint URI Response
Create a Product /product Created product as Mono
All products /products returns all products as Flux
Delate a product /product/{id} Mono
Update a product /product/{id} Updated product as Mono
Java
 




xxxxxxxxxx
1
59


 
1
package org.smarttechie.controller;
2
 
           
3
import org.smarttechie.model.Product;
4
import org.smarttechie.repository.ProductRepository;
5
import org.smarttechie.service.ProductService;
6
import org.springframework.beans.factory.annotation.Autowired;
7
import org.springframework.http.HttpStatus;
8
import org.springframework.http.MediaType;
9
import org.springframework.http.ResponseEntity;
10
import org.springframework.web.bind.annotation.*;
11
import reactor.core.publisher.Flux;
12
import reactor.core.publisher.Mono;
13
 
           
14
@RestController
15
public class ProductController {
16
 
           
17
    @Autowired
18
    private ProductService productService;
19
 
           
20
    /**
21
     * This endpoint allows to create a product.
22
     * @param product - to create
23
     * @return - the created product
24
     */
25
    @PostMapping("/product")
26
    @ResponseStatus(HttpStatus.CREATED)
27
    public Mono<Product> createProduct(@RequestBody Product product){
28
        return productService.save(product);
29
    }
30
 
           
31
    /**
32
     * This endpoint gives all the products
33
     * @return - the list of products available
34
     */
35
    @GetMapping("/products")
36
    public Flux<Product> getAllProducts(){
37
        return productService.getAllProducts();
38
    }
39
 
           
40
    /**
41
     * This endpoint allows to delete a product
42
     * @param id - to delete
43
     * @return
44
     */
45
    @DeleteMapping("/product/{id}")
46
    public Mono<Void> deleteProduct(@PathVariable int id){
47
        return productService.deleteProduct(id);
48
    }
49
 
           
50
    /**
51
     * This endpoint allows to update a product
52
     * @param product - to update
53
     * @return - the updated product
54
     */
55
    @PutMapping("product/{id}")
56
    public Mono<ResponseEntity<Product>> updateProduct(@RequestBody Product product){
57
        return productService.update(product);
58
    }
59
}



As we are building reactive APIs, we can build APIs with a functional style programming model without using RestController. In this case, we need to have a router and a handler component as shown below.

Java
 




xxxxxxxxxx
1
32


 
1
package org.smarttechie.router;
2
 
          
3
import org.smarttechie.handler.ProductHandler;
4
import org.springframework.context.annotation.Bean;
5
import org.springframework.context.annotation.Configuration;
6
import org.springframework.http.MediaType;
7
import org.springframework.web.reactive.function.server.RouterFunction;
8
import org.springframework.web.reactive.function.server.RouterFunctions;
9
import org.springframework.web.reactive.function.server.ServerResponse;
10
import static org.springframework.web.reactive.function.server.RequestPredicates.*;
11
@Configuration
12
public class ProductRouter {
13
 
          
14
    /**
15
     * The router configuration for the product handler.
16
     * @param productHandler
17
     * @return
18
     */
19
    @Bean
20
    public RouterFunction<ServerResponse> productsRoute(ProductHandler productHandler){
21
 
          
22
        return RouterFunctions
23
                .route(GET("/products").and(accept(MediaType.APPLICATION_JSON))
24
                        ,productHandler::getAllProducts)
25
                .andRoute(POST("/product").and(accept(MediaType.APPLICATION_JSON))
26
                        ,productHandler::createProduct)
27
                .andRoute(DELETE("/product/{id}").and(accept(MediaType.APPLICATION_JSON))
28
                        ,productHandler::deleteProduct)
29
                .andRoute(PUT("/product/{id}").and(accept(MediaType.APPLICATION_JSON))
30
                        ,productHandler::updateProduct);
31
    }
32
}


Java
 




xxxxxxxxxx
1
75


 
1
package org.smarttechie.handler;
2
 
          
3
import org.smarttechie.model.Product;
4
import org.smarttechie.service.ProductService;
5
import org.springframework.beans.factory.annotation.Autowired;
6
import org.springframework.http.MediaType;
7
import org.springframework.stereotype.Component;
8
import org.springframework.web.reactive.function.server.ServerRequest;
9
import org.springframework.web.reactive.function.server.ServerResponse;
10
import reactor.core.publisher.Mono;
11
import static org.springframework.web.reactive.function.BodyInserters.fromObject;
12
 
          
13
@Component
14
public class ProductHandler {
15
    @Autowired
16
    private ProductService productService;
17
    static Mono<ServerResponse> notFound = ServerResponse.notFound().build();
18
 
          
19
    /**
20
     * The handler to get all the available products.
21
     * @param serverRequest
22
     * @return - all the products info as part of ServerResponse
23
     */
24
    public Mono<ServerResponse> getAllProducts(ServerRequest serverRequest) {
25
 
          
26
        return ServerResponse.ok()
27
                .contentType(MediaType.APPLICATION_JSON)
28
                .body(productService.getAllProducts(), Product.class);
29
 
          
30
    }
31
 
          
32
    /**
33
     * The handler to create a product
34
     * @param serverRequest
35
     * @return - return the created product as part of ServerResponse
36
     */
37
    public Mono<ServerResponse> createProduct(ServerRequest serverRequest) {
38
 
          
39
        Mono<Product> productToSave = serverRequest.bodyToMono(Product.class);
40
 
          
41
        return productToSave.flatMap(product ->
42
                ServerResponse.ok()
43
                        .contentType(MediaType.APPLICATION_JSON)
44
                        .body(productService.save(product), Product.class));
45
 
          
46
    }
47
 
          
48
    /**
49
     * The handler to delete a product based on the product id.
50
     * @param serverRequest
51
     * @return - return the deleted product as part of ServerResponse
52
     */
53
    public Mono<ServerResponse> deleteProduct(ServerRequest serverRequest) {
54
 
          
55
        String id = serverRequest.pathVariable("id");
56
        Mono<Void> deleteItem = productService.deleteProduct(Integer.parseInt(id));
57
 
          
58
        return ServerResponse.ok()
59
                .contentType(MediaType.APPLICATION_JSON)
60
                .body(deleteItem, Void.class);
61
    }
62
 
          
63
    /**
64
     * The handler to update a product.
65
     * @param serverRequest
66
     * @return - The updated product as part of ServerResponse
67
     */
68
    public Mono<ServerResponse> updateProduct(ServerRequest serverRequest) {
69
        return productService.update(serverRequest.bodyToMono(Product.class)).flatMap(product ->
70
                ServerResponse.ok()
71
                        .contentType(MediaType.APPLICATION_JSON)
72
                        .body(fromObject(product)))
73
                .switchIfEmpty(notFound);
74
    }
75
}



So far, we have seen how to expose reactive REST APIs. With this implementation, I have done a simple benchmarking on reactive APIs versus the non-reactive APIs (built non-reactive APIs using Spring RestController) using Gatling. Below are the comparison metrics between the reactive and non-reactive APIs. This is not an extensive benchmarking. So, before adopting please make sure to do extensive benchmarking for your use case.

Load test results comparison

The Gatling load test scripts are also available on GitHub for your reference. With this, I conclude the article on “Build Reactive REST APIs with Spring WebFlux“. We will meet on another topic. Till then, Happy Learning!!

Topics:
java, project reactor, reactive streams, spring boot, spring web flux, tutorial

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}