Dilaver DEMİREL
Message-Driven Communication in a Microservices Architecture Using Spring Cloud Stream and RabbitMQ
In this article, I will attempt to create an example of message-based communication in distributed systems using Spring Cloud Stream and RabbitMQ. First, I’d like to touch on the theoretical aspects of the topic.
We can use synchronous or asynchronous methods for communication between applications. In distributed applications, a key consideration is avoiding tight coupling in inter-application communication.
For synchronous methods, we can use the HTTP request/response model as an example. Although HTTP requests can be processed asynchronously by the application using a non-blocking thread model when fulfilling the request, they are fundamentally a synchronous form of communication. This is because if the target service cannot respond immediately when the request is made, the requesting application cannot continue its process. For this reason, avoiding this type of communication as much as possible ensures the system maintains a robust character.
Message-based communication provides us with flexible connections between services. Even if the application receiving the message is not available at that moment, it will receive and respond to the request when it becomes available.
When we talk about asynchronous communication an alternative to this approach message-based communication is often the first thing that comes to mind. Message-based communication provides us with flexible connections between services. Even if the application receiving the message is not available at that moment, it will receive and respond to the request when it becomes available.
This flexibility necessitates the inclusion of different components in the application stack. For example, we need to include a tool like RabbitMQ, which manages, buffers, and handles errors in the message traffic within the cluster. While every new component added to the system is included to solve problems, it also incurs management and maintenance costs. Depending on the application’s characteristics, solving the business problem with the fewest possible components ensures we have a simpler and more sustainable system.
- While this approach offers advantages for solving critical business problems, it also presents the following situations that require management and attention:
- To ensure the application’s reliability, it is crucial that this tool does not create a bottleneck. If a tool like RabbitMQ is used, it must be configured as a cluster. Having a redundant component will make the system more robust.
- If the operations performed during message transmission occur within a database transaction, the message must be sent within the integrity of that transaction. A message may have been sent while the operations are in progress, but if an error occurs during the commit process at the database level, the data cannot be written to the database. However, the message will have already been sent.
- During message reception/consumption, a message may fail to be processed due to an error. It may even fail to be processed after several attempts. In this case, the undelivered message must be managed.
- Messages may need to be processed in a specific order.
- The same message may have been sent twice for some reason.
As we can see, there are important factors that affect the success, quality, and robustness of the application.
Spring Cloud Stream offers the following benefits in this regard: as stated in its definition, it is a framework that enables the development of highly scalable microservices applications.
Its greatest advantage is that it abstracts away many message queuing systems such as RabbitMQ and Kafka which it refers to as “binders.” This means that when developing our application, we build it using Spring Cloud Stream’s APIs rather than the APIs of systems like RabbitMQ or Kafka. As a result, our application can work with RabbitMQ today and Apache Kafka tomorrow.
Now let’s look at how we can develop a system using this approach. In this example, as mentioned earlier, I will not be using Spring Cloud Stream and RabbitMQ.
Sample Application
In the scenario I'll be implementing, there are two services: PaymentService and StockService. The PaymentService processes requests received via the REST API, processes the payment, and, once the transaction is complete, sends a request to the StockService to initiate the shipment.
Dependencies
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
StockService – application.yml
spring:
cloud:
stream:
bindings:
stockOperationInputChannel:
destination: stock-service.operation
group: stock-service-group
binder: local_rabbit
rabbit:
bindings:
stockOperationInputChannel:
consumer:
autoBindDlq: true
republishToDlq: true
binders:
local_rabbit:
type: rabbit
environment:
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
The configuration is divided into three sections under the `stream` section: `bindings`, `rabbit`, and `binders`.
`Bindings` allows you to define channels that represent queue/topic/exchange structures for messages within Spring. The `destination` element under `bindings` allows you to specify the queue definition on the message service.
The `group` configuration ensures that messages are processed in a load-balanced manner on the consumer side. By default, clients or consumers listening to a queue in RabbitMQ each receive every message that arrives in the queue. This is generally undesirable because, when there are multiple consumer instances, all consumers process the same message. This results in redundant processing of the same task. Spring Cloud Stream resolves this issue using the `group` configuration. Consumers within the same group process messages in sequence, ensuring load balancing and transaction consistency.
The `binder` specifies which service (RabbitMQ, Apache Kafka) it will run on, as defined in the `binders` section of the configuration.
The Rabbit section is where parameters specific to RabbitMQ are defined, in addition to the abstraction provided by Spring Cloud Stream. Here, you can customize the relevant channel. The parameters we define here are used to create a queue known as the “Dead Letter Queue (DLQ),” which is designed to resolve the issue of “messages not being consumed or received” mentioned earlier. What is done here is that if a message cannot be delivered or consumed after three default attempts, it is routed to the queue specified as the DLQ. This prevents problematic messages from being lost. On the consumer side, these problematic messages can be captured by a listener, logged, corrected, and republished, or managed in similar ways.
Executing the Scenario
1 – Receiving the payment request (Payment Service)
2 – Processing the payment and sending a message to the inventory service (Payment Service)
As mentioned above, a transactional operation is taking place here. The message required for inventory operations must not be sent to the inventory service until the transaction is successfully completed. To solve this problem, we use Spring’s ApplicationEventPublisher. ApplicationEventPublisher allows us to publish events within the application context. Another advantage it offers is that it allows us to make this process part of the transaction.
As seen above, two annotations allow us to listen to events. The TransactionalEventListener, on the other hand, allows the message to be processed after a transaction or in other provided stages if a transaction is ongoing at the time the message is captured. We do this in the AFTER_COMMIT stage because the successful completion of the transaction is crucial for us.
But what if the transaction is successfully completed and the message is now to be sent to the stock service? However, RabbitMQ is not accessible at that moment. In this case, the message cannot be sent, and transaction integrity will be compromised. We can solve this problem with the Outbox Pattern. I didn't implement it in this example, but briefly, the solution is as follows: messages are saved to a database table, and then, through a job, messages that cannot be sent are delivered to RabbitMQ in a controlled manner. This prevents message loss and ensures transaction integrity.
3 – Processing messages received by StockService;
Now we can look at how these definitions we specified in the configuration can be accessed and used from within the application. For this, two definitions need to be made within the application, the first is the Channel definition;
Channel Definition
The second is the listener definition, which enables the reception of messages.
Listener Definition
Messages from other services are received and the process continues. On the other hand, messages that cannot be received are logged using the DEAD_LETTER_QUEUE listener. To trigger a dead letter, I've enabled an error to be thrown if the price drops to -1. This will create a dead letter.
Finally, for the Channel definition to be active within the application, the following definition (EnableBinding) must be made.
Binding Activation
RabbitMQ
I tried to create a perspective on how to implement a message-driven communication approach in a more manageable way in distributed architectures like microservices.