Event-driven design with Spring Reactive Redis Pub/Sub model
![](https://cdn-images-1.readmedium.com/v2/resize:fit:800/1*fOWMIT4fMZce6PH2_na4Rg.png)
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.
You can refer to my other articles related to reactive architecture using reactive Redis down below.
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.
![](https://cdn-images-1.readmedium.com/v2/resize:fit:800/1*TdfNz9LBh49If16KoMESqg.png)
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
RedisMessageListenerContainer
acts 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.
![](https://cdn-images-1.readmedium.com/v2/resize:fit:800/1*NQO2OITjBBUgZxmwrB6j2g.png)
Redis Insight Dashboard with Published Products
![](https://cdn-images-1.readmedium.com/v2/resize:fit:800/1*KKfvBmkvgV0AD_s4NaswQg.png)
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.