Microservice Communication: A Spring Integration Tutorial With Redis

Microservice architecture is a very popular approach in designing and implementing highly scalable web applications. Communication within a monolithic application between components is usually based on method or function calls within the same process. A microservices‑based application, on the other hand, is a distributed system running on multiple machines. Communication between these microservices is important in order to have a stable and scalable system. There are multiple ways to do this. Message-based communication is one way to do this reliably.

When using messaging, components interact with each other by asynchronously exchanging messages. Messages are exchanged through channels.

When Service A wants to communicate with Service B, instead of sending it directly, A sends it to a specific channel. When Service B wants to read the message, it picks up the message from a particular message channel.

In this Spring Integration tutorial, you will learn how to implement messaging in a Spring application using Redis. The guide includes a walk-through of an example application where one service is pushing events in the queue and another service is processing these events one by one.

Spring Integration

The project Spring Integration extends the Spring framework to provide support for messaging between or within Spring-based applications. Components are wired together via the messaging paradigm. Individual components may not be aware of other components in the application.

Spring Integration provides a wide selection of mechanisms to communicate with external systems. Channel adapters are one such mechanism used for one-way integration (send or receive). And gateways are used for request/reply scenarios (inbound or outbound).

Apache Camel is an alternative that is widely used. Spring Integration is usually preferred in existing Spring-based services as it is part of the Spring ecosystem.

Redis

Redis is an extremely fast in-memory data store. It can optionally persist to a disk also. It supports different data structures like simple key-value pairs, sets, queues, etc. Using Redis as a queue makes sharing data between components and horizontal scaling much easier. A producer or multiple producers can push data to the queue, and a consumer or multiple consumers can pull the data and process the event. Multiple consumers cannot consume the same event—this ensures that one event is processed once.

Benefits of using Redis as a message queue:

• Parallel execution of discrete tasks in a non-blocking fashion
• Great performance
• Stability
• Easy monitoring and debugging
• Easy implementation and usage

Rules:

• Adding a task to the queue should be faster than processing the task itself.
• Consuming tasks should be faster than producing them (and if not, add more consumers).

Spring Integration with Redis

The following walks through the creation of a sample application to explain how to use Spring Integration with Redis.

Let’s say you have an application which allows users to publish posts. And you want to build a follow feature. Another requirement is that every time someone publishes a post, all followers should be notified via some communication channel (e.g., email or push notification).

One way to implement this is to send an email to each follower once the user publishes something. But what happens when the user has 1,000 followers? And when 1,000 users publish something in 10 seconds, each one of whom has 1,000 followers? Also, will the publisher’s post wait until all emails are sent?

Distributed systems resolve this problem.

This specific problem could be resolved by using a queue. Service A (the producer), which is responsible for publishing posts, will do just that. It will publish a post and push an event with the list of users who need to receive an email and the post itself. The list of users could be fetched in service B, but for simplicity of this example, we will send it from service A. This is an asynchronous operation. This means the service that is publishing will not have to wait to send emails.

Service B (the consumer) will pull the event from the queue and process it. This way, we could easily scale our services, and we could have n consumers sending emails (processing events).

So let’s start with an implementation in the producer’s service. Necessary dependencies are:

<dependency>
   <groupId>redis.clients</groupId>
   <artifactId>jedis</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.data</groupId>
   <artifactId>spring-data-redis</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.integration</groupId>
   <artifactId>spring-integration-redis</artifactId>
</dependency>

These three Maven dependencies are necessary:

• Jedis is a Redis client.
• The Spring Data Redis dependency makes it easier to use Redis in Java. It provides familiar Spring concepts such as a template class for core API usage and lightweight repository-style data access.
• Spring Integration Redis provides an extension of the Spring programming model to support the well-known Enterprise Integration Patterns.

Next, we need to configure the Jedis client:

@Configuration
public class RedisConfig {

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

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

   @Bean
   public JedisPoolConfig poolConfig() {
       JedisPoolConfig poolConfig = new JedisPoolConfig();
       poolConfig.setMaxTotal(128);
       return poolConfig;
   }

   @Bean
   public RedisConnectionFactory redisConnectionFactory(JedisPoolConfig poolConfig) {
       final JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
       connectionFactory.setHostName(redisHost);
       connectionFactory.setPort(redisPort);
       connectionFactory.setPoolConfig(poolConfig);
       connectionFactory.setUsePool(true);
       return connectionFactory;
   }
}

The annotation @Value means that Spring will inject the value defined in the application properties into the field. This means redis.host and redis.port values should be defined in the application properties.

Now, we need to define the message we want to send to the queue. A simple example message could look like:

@Getter
@Setter
@Builder
public class PostPublishedEvent {
   
   private String postUrl;
   private String postTitle;    
   private List<String> emails;

}

Note: (Project Lombok) provides the @Getter, @Setter, @Builder, and many other annotations to avoid cluttering code with getters, setters, and other trivial stuff. Project Lombok is an effective tool for streamlining your Java code.

The message itself will be saved in JSON format in the queue. Every time an event is published to the queue, the message will be serialized to JSON. And when consuming from the queue, the message will be deserialized.

With the message defined, we need to define the queue itself. In Spring Integration, it can be easily done via an .xml configuration. The configuration should be placed inside the resources/WEB-INF directory.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:int="http://www.springframework.org/schema/integration"
      xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
      xsi:schemaLocation="http://www.springframework.org/schema/integration/redis
      http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
     http://www.springframework.org/schema/integration
     http://www.springframework.org/schema/integration/spring-integration.xsd
     http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans.xsd">

   <int-redis:queue-outbound-channel-adapter
           id="event-outbound-channel-adapter"
           channel="eventChannelJson"
           serializer="serializer"
           auto-startup="true" connection-factory="redisConnectionFactory"
           queue="my-event-queue" />

   <int:gateway id="eventChannelGateway"
                service-interface="org.toptal.queue.RedisChannelGateway"
                error-channel="errorChannel" default-request-channel="eventChannel">
       <int:default-header name="topic" value="queue"/>
   </int:gateway>

   <int:channel id="eventChannelJson"/>
   <int:channel id="eventChannel"/>
   

   <bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>

   <int:object-to-json-transformer input-channel="eventChannel"
                                   output-channel="eventChannelJson"/>

</beans>

In the configuration, you can see int-redis:queue-outbound-channel-adapter. Its properties are:

• id: The bean name of the component.
• channel: MessageChannel from which this endpoint receives messages.
• connection-factory: A reference to a RedisConnectionFactory bean.
• queue: The name of the Redis list on which the queue-based push operation is performed to send Redis messages. This attribute is mutually exclusive with queue-expression.
• queue-expression: A SpEL expression to determine the name of the Redis list using the incoming message at runtime as the #root variable. This attribute is mutually exclusive with the queue.
• serializer: A RedisSerializer bean reference. By default, it is a JdkSerializationRedisSerializer. However, for String payloads, a StringRedisSerializer is used if a serializer reference isn’t provided.
• extract-payload: Specify if this endpoint should send just the payload to the Redis queue or the entire message. Its default value is true.
• left-push: Specify whether this endpoint should use left push (when true) or right push (when false) to write messages to the Redis list. If true, the Redis list acts as a FIFO queue when used with a default Redis queue inbound channel adapter. Set to false to use with software that reads from the list with left pop or to achieve a stack-like message order. Its default value is true.

The next step is to define the gateway, which is mentioned in the .xml configuration. For a gateway, we are using the RedisChannelGateway class from the org.toptal.queue package.

StringRedisSerializer is used to serialize message before saving in Redis. Also in the .xml configuration, we defined the gateway and set RedisChannelGateway as a gateway service. This means that the RedisChannelGateway bean could be injected into other beans. We defined the property default-request-channel because it’s also possible to provide per-method channel references by using the @Gateway annotation. Class definition:

public interface RedisChannelGateway {
   void enqueue(PostPublishedEvent event);
}

To wire this configuration into our application, we have to import it. This is implemented in the SpringIntegrationConfig class.

@ImportResource("classpath:WEB-INF/event-queue-config.xml")
@AutoConfigureAfter(RedisConfig.class)
@Configuration
public class SpringIntegrationConfig {
}

@ImportResource annotation is used to import Spring .xml configuration files into @Configuration. And @AutoConfigureAfter annotation is used to hint that an auto-configuration should be applied after other specified auto-configuration classes.

We will now create a service and implement the method that will enqueue events to the Redis queue.

public interface QueueService {

   void enqueue(PostPublishedEvent event);
}

@Service
public class RedisQueueService implements QueueService {

   private RedisChannelGateway channelGateway;

   @Autowired
   public RedisQueueService(RedisChannelGateway channelGateway) {
       this.channelGateway = channelGateway;
   }

   @Override
   public void enqueue(PostPublishedEvent event) {
       channelGateway.enqueue(event);
   }
}

And now, you can easily send a message to the queue using the enqueue method from QueueService.

Redis queues are simply lists with one or more producer and consumer. To publish a message to a queue, producers use the LPUSH Redis command. And if you monitor Redis (hint: type redis-cli monitor), you can see that the message is added to the queue:

"LPUSH" "my-event-queue" "{\"postUrl\":\"test\",\"postTitle\":\"test\",\"emails\":[\"test\"]}"

Now, we need to create a consumer application which will pull these events from the queue and process them. The consumer service needs the same dependencies as the producer service.

Now we can reuse the PostPublishedEvent class to deserialize messages.

We need to create the queue config and, again, it has to be placed inside the resources/WEB-INF directory. The content of the queue config is:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:int="http://www.springframework.org/schema/integration"
      xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
      xsi:schemaLocation="http://www.springframework.org/schema/integration/redis
      http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
     http://www.springframework.org/schema/integration
     http://www.springframework.org/schema/integration/spring-integration.xsd
     http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans.xsd">

   <int-redis:queue-inbound-channel-adapter id="event-inbound-channel-adapter"
                                            channel="eventChannelJson" queue="my-event-queue"
                                            serializer="serializer" auto-startup="true"
                                            connection-factory="redisConnectionFactory"/>

   <int:channel id="eventChannelJson"/>

   <int:channel id="eventChannel">
       <int:queue/>
   </int:channel>

   <bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>

   <int:json-to-object-transformer input-channel="eventChannelJson"
                                   output-channel="eventChannel"
                                   type="com.toptal.integration.spring.model.PostPublishedEvent"/>

   <int:service-activator input-channel="eventChannel" ref="RedisEventProcessingService"
                          method="process">
       <int:poller fixed-delay="10" time-unit="SECONDS" max-messages-per-poll="500"/>
   </int:service-activator>

</beans>

In the .xml configuration, int-redis:queue-inbound-channel-adapter can have the following properties:

• id: The bean name of the component.
• channel: The MessageChannel to which we send messages from this endpoint.
• auto-startup: A SmartLifecycle attribute to specify whether this endpoint should start automatically after the application context start or not. Its default value is true.
• phase: A SmartLifecycle attribute to specify the phase in which this endpoint will be started. Its default value is 0.
• connection-factory: A reference to a RedisConnectionFactory bean.
• queue: The name of the Redis list on which the queue-based pop operation is performed to get Redis messages.
• error-channel: The MessageChannel to which we will send ErrorMessages with Exceptions from the listening task of the Endpoint. By default, the underlying MessagePublishingErrorHandler uses the default errorChannel from the application context.
• serializer: The RedisSerializer bean reference. It can be an empty string, which means no serializer. In this case, the raw byte[] from the inbound Redis message is sent to the channel as the Message payload. By default, it is a JdkSerializationRedisSerializer.
• receive-timeout: The timeout in milliseconds for the pop operation to wait for a Redis message from the queue. Its default value is 1 second.
• recovery-interval: The time in milliseconds for which the listener task should sleep after exceptions on the pop operation before restarting the listener task.
• expect-message: Specify if this endpoint expects data from the Redis queue to contain entire messages. If this attribute is set to true, the serializer can’t be an empty string because messages require some form of deserialization (JDK serialization by default). Its default value is false.
• task-executor: A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) bean. It is used for the underlying listening task. By default, a SimpleAsyncTaskExecutor is used.
• right-pop: Specify whether this endpoint should use right pop (when true) or left pop (when false) to read messages from the Redis list. If true, the Redis list acts as a FIFO queue when used with a default Redis queue outbound channel adapter. Set to false to use with software that writes to the list with right push or to achieve a stack-like message order. Its default value is true.

The important part is the service-activator, which defines which service and method should be used to process the event. Also, the json-to-object-transformer needs a type attribute in order to transform JSON to objects, set above to type="com.toptal.integration.spring.model.PostPublishedEvent".

Again, to wire this config, we will need the SpringIntegrationConfig class, which can be the same as before. And lastly, we need a service which will actually process the event.

public interface EventProcessingService {
   void process(PostPublishedEvent event);
}

@Service("RedisEventProcessingService")
public class RedisEventProcessingService implements EventProcessingService {

   @Override
   public void process(PostPublishedEvent event) {
       // TODO: Send emails here, retry strategy, etc :)
   }

}

Once you run the application, you can see in Redis:

"BRPOP" "my-event-queue" "1"

Conclusion

With Spring Integration and Redis, building a Spring microservices application is not as daunting as it normally would be. With a little configuration and a small amount of boilerplate code, you can build the foundations of your microservice architecture in no time.

Even if you do not plan to scratch your current Spring project entirely and switch to a new architecture, with the help of Redis, it is very simple to gain huge performance improvements with queues.