Reactive Programming
Reactive programming is a programming paradigm that manages asynchronous data streams and automatically propagates changes, enabling systems to react to events in real time. It’s useful for creating responsive APIs and event-driven applications, often applied in UI updates, data streams, and real-time systems.
WebFlux
WebFlux is designed for applications with high concurrency needs. It leverages Project Reactor and Reactive Streams, enabling it to handle a large number of requests concurrently with minimal resource usage.
Key Features
- Reactive programming uses reactive types like
Mono, which manages data 0..1, andFlux, which manages data 0..N, to process data streams asynchronously. - Non-blocking I/O is built on non-blocking servers like Netty, reducing overhead and allowing for high-throughput processing.
- Functional and annotation-based models support both functional routing and traditional annotation-based controllers.
R2DBC (Reactive Database Connectivity)
R2DBC (Reactive Relational Database Connectivity) is a non-blocking API for interacting with relational databases in a fully reactive, asynchronous manner. It’s designed for reactive applications, enabling efficient handling of database operations in real-time data streams, especially with frameworks like Spring WebFlux.
Overview of the Solution
This approach is ideal for applications that require scalable, real-time interactions with data sources. Spring WebFlux allows for a non-blocking, asynchronous API setup, while R2DBC provides reactive connections to relational databases, like PostgreSQL or MySQL, which traditionally require blocking I/O with JDBC. This combination allows for a seamless, event-driven system, leveraging reactive streams to handle data flow from the client to the database without waiting for blocking operations.
Key Components
- Java 17 or later will execute this solution because of the used Java record.
- Spring Boot library is used as the portable configuration of the services.
- Spring APO declarative approach means concerns are added dynamically at runtime, not hard coded into the application, making adjustments easier without altering existing code.
- Spring WebFlux: A reactive web framework that provides fully
non-blockingAPIs, optimized for reactive applications and real-time web functionality - R2DBC: A reactive, non-blocking API for database connectivity, which replaces the
blockingbehavior of JDBC with a fully reactive stack - MapStruct: Used to transform requests into the data access layer
WebFlux Servlet
WebFlux supports a variety of servers like Netty, Tomcat, Jetty, Undertow, and Servlet containers. Let’s define Netty and servlet container in this example.
Project Structure
Configuration
logging:
level:
org:
springframework:
r2dbc: DEBUG
server:
error:
include-stacktrace: never
spring:
application:
name: async-api-async-db
jackson:
serialization:
FAIL_ON_EMPTY_BEANS: false
main:
allow-bean-definition-overriding: true
r2dbc:
password: 123456
pool:
enabled: true
initial-size: 5
max-idle-time: 30m
max-size: 20
validation-query: SELECT 1
url: r2dbc:mysql://localhost:3306/test
username: rootWeb Flux Netty: Functional Process
Spring’s functional web framework (ProductRoute.java) exposes routing functionality, such as creating a RouterFunction using a discoverable builder-style API, to create a RouterFunction given a RequestPredicate and HandlerFunction, and to do further subrouting on an existing routing function. Additionally, this class can transform a RouterFunction into an HttpHandle.
package com.amran.async.controller.functional;
import com.amran.async.constant.ProductAPI;
import com.amran.async.handler.ProductHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;
import static org.springframework.web.reactive.function.server.RequestPredicates.DELETE;
import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RequestPredicates.POST;
import static org.springframework.web.reactive.function.server.RequestPredicates.PUT;
/**
* @author Md Amran Hossain on 28/10/2024 AD
* @Project async-api-async-db
*/
@Configuration(proxyBeanMethods = false)
public class ProductRoute {
@Bean
public RouterFunction<ServerResponse> routerFunction(ProductHandler productHandler) {
return RouterFunctions
.route(GET(ProductAPI.GET_PRODUCTS).and(ProductAPI.ACCEPT_JSON), productHandler::getAllProducts)
.andRoute(GET(ProductAPI.GET_PRODUCT_BY_ID).and(ProductAPI.ACCEPT_JSON), productHandler::getProductById)
.andRoute(POST(ProductAPI.ADD_PRODUCT).and(ProductAPI.ACCEPT_JSON), productHandler::handleRequest)
.andRoute(DELETE(ProductAPI.DELETE_PRODUCT).and(ProductAPI.ACCEPT_JSON), productHandler::deleteProduct)
.andRoute(PUT(ProductAPI.UPDATE_PRODUCT).and(ProductAPI.ACCEPT_JSON), productHandler::handleRequest);
}
}
WebFlux Non-Functional Request Process
- Spring Detected Rest API (
ProductController.java):
package com.amran.async.controller;
import com.amran.async.model.Product;
import com.amran.async.service.ProductService;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* @author Md Amran Hossain on 28/10/2024 AD
* @Project async-api-async-db
*/
@RestController
@RequestMapping(path = "/api/v2")
public class ProductController {
private final ProductService productService;
public ProductController(ProductService productService) {
this.productService = productService;
}
@GetMapping("/product")
@ResponseStatus(HttpStatus.OK)
public Flux<Product> getAllProducts() {
return productService.getAllProducts();
}
@GetMapping("/product/{id}")
@ResponseStatus(HttpStatus.OK)
public Mono<Product> getProductById(@PathVariable("id") Long id) {
return productService.getProductById(id);
}
@PostMapping("/product")
@ResponseStatus(HttpStatus.CREATED)
public Mono<Product> createProduct(@RequestBody Product product) {
return productService.addProduct(product);
}
@PutMapping("/product/{id}")
@ResponseStatus(HttpStatus.OK)
public Mono<Product> updateProduct(@PathVariable("id") Long id, @RequestBody Product product) {
return productService.updateProduct(product, id);
}
@DeleteMapping("/product/{id}")
@ResponseStatus(HttpStatus.NO_CONTENT)
public Mono<Void> deleteProduct(@PathVariable("id") Long id) {
return productService.deleteProduct(id);
}
}
Next, let’s turn to see the R2DBC implementation. This repository follows reactive paradigms and uses Project Reactor types which are built on top of Reactive Streams. Save and delete operations with entities that have a version attribute trigger an onError with an OptimisticLockingFailureException when they encounter a different version value in the persistence store than in the entity passed as an argument. Other delete operations that only receive IDs or entities without version attributes do not trigger an error when no matching data is found in the persistence store.
Optimistic Lock
Optimistic locking and a concurrency conflict are detected while updating data. Optimistic locking is a mechanism for handling concurrent modifications in a way that ensures data integrity without requiring exclusive database locks. This is commonly applied in systems where multiple transactions might read and update the same record concurrently.
package com.amran.async.repository;
import com.amran.async.domain.ProductEntity;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
/**
* @author Md Amran Hossain on 28/10/2024 AD
* @Project async-api-async-db
*/
@Repository
public interface ProductRepository extends ReactiveCrudRepository<ProductEntity, Long> {
// @Query("SELECT * FROM product WHERE product_name = :productName")
// Flux<ProductEntity> findByProductName(String productName);
}Summary
In summary, using Java Spring Boot WebFlux with R2DBC allows developers to build high-performance, reactive REST APIs with non-blocking database connections. This combination supports scalable, low-latency applications optimized for handling large volumes of concurrent requests, making it ideal for real-time, cloud-native environments.
Also, you guys can find a full source code example here.
