Summary

The website content provides a comprehensive guide on implementing an event-driven design using the Reactive Redis Publish/Subscribe model within a Spring Boot application.

Abstract

The article delves into the design of an event-driven architecture utilizing the Reactive Redis Publish/Subscribe model. It begins with an overview of Reactive Redis, referencing previous work on integrating it with Spring WebFlux. The author then explains the Redis Publisher/Subscriber model, emphasizing its ability to facilitate anonymous communication between components in a distributed system. The practical setup includes using Docker to run Redis with Redis Insight, configuring a Spring Boot application with the necessary dependencies, and setting up Reactive Redis Pub/Sub. The configuration and implementation details are provided for the RedisConfig, ProductPublisher, ProductService, and ProductController classes, showcasing how to publish and consume messages. The article concludes with instructions on testing the application using Spring Doc Open API and visualizing the results in Redis Insight.

Opinions

The author suggests that the Reactive Redis Publish/Subscribe model is suitable for designing event-driven architectures.
The article implies that using Docker for starting Redis and Redis Insight simplifies the setup process.
The author emphasizes the importance of Reactive Redis operations for handling asynchronous data flow, which is a key aspect of reactive programming.
The use of Spring Boot's @Bean annotation for configuration is presented as a clean approach to managing Redis connections and operations.
The article promotes the Spring ecosystem, including Spring WebFlux and Spring Doc Open API, for building reactive applications and testing REST endpoints.
The inclusion of previous articles on related topics suggests that the author values comprehensive learning resources for developers.
The author encourages reader feedback, indicating a willingness to engage with the community and improve upon the shared knowledge.

Event-driven design with Spring Reactive Redis Pub/Sub model

In this article, I will talk about designing an event-driven architecture with the Reactive Redis Publish/Subscribe model.

Overview of the Reactive Redis

I have already discussed a detailed article about how to integrate Reactive Redis with Spring WebFlux. You can refer to it below.

Building a Reactive Architecture with Spring WebFlux and Reactive Redis

In this article, I will talk about how to design a reactive application with Spring WebFlux and Reactive Redis.

blog.stackademic.com

You can refer to my other articles related to reactive architecture using reactive Redis down below.

How to handle a write-heavy workload effectively with Spring Reactive Redis

Building a Reactive Architecture to handle write-heavy workloads around Spring Reactive Redis

blog.stackademic.com

Redis Publisher/Subscriber Model

Pub/Sub is a messaging model that allows different components in a distributed system to communicate with one another. Publishers send messages to a topic, and subscribers receive messages from that topic, allowing publishers to send messages to subscribers while remaining anonymous. The Pub/Sub system ensures that the message reaches all subscribers who are interested in the topic.

One distinguishable aspect of the Redis Pub/Sub model is that messages published by the publishers to the channels will be pushed by Redis to all the subscribed clients. Subscribers receive the messages in the order that they are published.

Application Setup with Spring Boot and Reactive Redis

Project Technical Stack

Spring Boot 3.2.0
Java 21
Redis
Docker

Starting Up Redis Server and Redis Insight

We will use Docker to start Redis with Redis Insight. The port of our Redis server is 6379, and the port of Redis Insight is 8001.

docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

We will then set up the application with Spring Initializer with the following dependencies in the pom.xml file.

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

<dependency>
   <groupId>org.projectlombok</groupId>
   <artifactId>lombok</artifactId>
   <optional>true</optional>
</dependency>

<dependency>
   <groupId>io.projectreactor</groupId>
   <artifactId>reactor-test</artifactId>
   <scope>test</scope>
</dependency>

<dependency>
   <groupId>org.springdoc</groupId>
   <artifactId>springdoc-openapi-starter-webflux-ui</artifactId>
   <version>2.3.0</version>
</dependency>

Setting Up Configurations

Configure theapplication.yml file for the SpringBoot application as follows: We will define the configurations related to Redis inside it.

redis:
  host: localhost
  port: 6379

Reactive Redis Pub/Sub Configuration

RedisMessageListenerContaineracts as a message-listener container. It is used to receive messages from a Redis channel and drive the MessageListener instances that are injected into it.

@Configuration
@Slf4j
public class RedisConfig {

    @Value("${redis.host}")
    private String host;

    @Value("${redis.port}")
    private int port;

    @Bean
    @Primary
    public ReactiveRedisConnectionFactory reactiveRedisConnectionFactory() {
        return new LettuceConnectionFactory(host, port);
    }

    @Bean
    public ReactiveRedisOperations<String, Product> reactiveRedisOperations(ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
        Jackson2JsonRedisSerializer<Product> serializer = new Jackson2JsonRedisSerializer<>(Product.class);

        RedisSerializationContext.RedisSerializationContextBuilder<String, Product> builder =
                RedisSerializationContext.newSerializationContext(new StringRedisSerializer());

        RedisSerializationContext<String, Product> context = builder.value(serializer).build();

        return new ReactiveRedisTemplate<>(reactiveRedisConnectionFactory, context);
    }

    @Bean
    public ReactiveRedisMessageListenerContainer messageListenerContainer(final ProductService productService, final ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
        final ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(reactiveRedisConnectionFactory);
        final ObjectMapper objectMapper = new ObjectMapper();
        container.receive(ChannelTopic.of("products"))
                .map(ReactiveSubscription.Message::getMessage)
                .map(message -> {
                    try {
                        return objectMapper.readValue(message, Product.class);
                    } catch (IOException e) {
                        return null;
                    }
                })
                .switchIfEmpty(Mono.error(new IllegalArgumentException()))
                .flatMap(productService::save)
                .subscribe(c-> log.info("Product Saved Successfully."));
        return container;
    }

}

Product Publisher Class

The Product Publisher class acts as the Redis Publisher, publishing to the desired channel.

@Component
@RequiredArgsConstructor
@Slf4j
public class ProductPublisher {

    private static final String REDIS_CHANNEL = "PRODUCT_CHANNEL";
    private final ReactiveRedisOperations<String, Product> reactiveRedisOperations;

    public Mono<String> publishProductMessage(final Product product) {
        this.reactiveRedisOperations.convertAndSend(REDIS_CHANNEL, product).subscribe(count -> {
            log.info("Product Published Successfully.");
        });
        return Mono.just("Published Message to Redis Channel");
    }

}

Product Service Class

@Service
@RequiredArgsConstructor
public class ProductService {

    private final ReactiveRedisOperations<String, Product> reactiveRedisOperations;
    private static final String REDIS_KEY = "PRODUCT:REDIS";


    public Flux<Product> findAll() {
        return this.reactiveRedisOperations.opsForList().range(REDIS_KEY, 0, -1);
    }

    public Mono<Long> save(final Product product) {
        final String id = UUID.randomUUID().toString().substring(0, 8);
        product.setId(id);
        return this.reactiveRedisOperations.opsForList().rightPush(REDIS_KEY, product);
    }

}

Product Controller Class

@RestController
@RequestMapping("/product")
@RequiredArgsConstructor
public class ProductController {

    private final ProductService productService;
    private final ProductPublisher productPublisher;

    @GetMapping
    public Flux<Product> findAll(){
        return this.productService.findAll();
    }

    @PostMapping("/publish")
    public Mono<String> publishProduct(final Product product){
        return this.productPublisher.publishProductMessage(product);
    }

}

Testing the Application

We will use the Spring Doc Open API to publish product messages to the Redis Channel.

Publish Product Message to Redis Channel

Redis Insight Dashboard with Published Products

Summary

In this article, we have covered:

Overview of Reactive Redis
Redis Publisher Subscriber model
How to start a Redis Server and Redis Insight with Docker
Reactive Redis Pub/Sub Integration with Spring Boot.
Testing the Rest Endpoints using Spring Doc Open API

Please feel free to share your feedback.

Thanks for reading.