Introduction
In our previous articles, we have discussed the REST standard primarily when discussing micro-service communication styles. Message-driven communication is an alternative that guarantees asynchronous, highly decoupled interaction. This article will teach you how to implement message-driven services using Spring Cloud.
Message-Driven Services
We can distinguish two main different patterns based on messages:
- Point-to-point
- Publish-subscribe
Point-to-point involves exchanging messages between two established endpoints. In the public-subscribe model, on the other hand, messages are sent to a so-called topic. Every service that wants to consume those messages must register as a listener to a topic. This article will focus on the publish-subscribe model, i.e. the best way to reduce coupling between components.
The basic components of messaging-based systems are producers and consumers. A producer sends messages created by some source to a remote destination (a topic), and a consumer registers to the same destination to receive the messages.
The exchange of messages is done through specialized software components called message brokers. Examples are Rabbit MQ and Kafka. We can integrate micro-services based on Spring Cloud Stream and message brokers through the so-called binders. A binder provides a broker-specific implementation of the Spring Cloud Stream abstraction.
Once we configure a specific binder, we have established an integration between our Spring Cloud Stream application and a broker. Using Spring Cloud Stream and broker-specific configurations, we can define the basic connection properties and the bindings between our application and broker topics, that we can call destinations.
Message-Driven Style Broker Running and Configuration
To have a functioning sample we must have a running message broker. If we use RabbitMQ, we can run it by using docker, with a docker-compose file like this:
version: '3'
services:
rabbitmq:
image: rabbitmq:management
ports:
- 5672:5672
- 15672:15672
To start it we have to position ourselves in the folder where the docker-compose file is and enter the command “./docker compose up” in the command line. Once we have RabbitMS running, we can configure it in our application.xml file:
spring:
application:
name: booksmanagement-service
rabbitmq:
host: localhost
port: 5672
Message-Driven Style Implemented by Java Functional Interfaces
In the recent versions of Spring Cloud Stream, message-driven services can use Java functional interfaces to implicitly implement message producers and consumers. Three interfaces are used:
- Consumer<T>, where T is the type of input
- Supplier<T>, where T is the type of input
- Function<T, R>, where T is the type of input and R is the type of the result
The Consumer interface, as its name implies, is implemented to provide a consumer in the publish-subscribe pattern, while a Supplier object implements a producer. The Function interface is a mixing of the two, it implements both a producer and a consumer in a single object. It connects to the broker to send a message to a topic and at the same time returns a value to an output topic.
To provide implementations of those three interfaces to obtain implicit producers and consumers we define Spring beans in the application, as in the following example:
@SpringBootApplication
public class AppMain {
public static void main(String[] args) {
new SpringApplicationBuilder(AppMain.class).run(args);
}
@Bean
public Consumer<String> sink() {
return payload -> {
System.out.println(payload);
};
}
@Bean
public Supplier<String> source() {
return () -> {
String message = "Supplier message";
System.out.println("Sending value: " + message);
return message;
};
}
@Bean
public Function<String, String> uppercase() {
return payload -> {
System.out.println(payload);
return payload.toUpperCase();
};
}
}
The name of the bean translates automatically to a binding with the format <function-name>-in-<index> for consumers and <function-name>-out-<index> for producers. The index part is 0 by default and has a particular significance only in the case of functions with multiple input/output arguments.
We can map the function names to remote destinations (topics or queues), like in the following piece of configuration:
spring:
application:
name: booksmanagement-service
rabbitmq:
host: localhost
port: 5672
cloud:
function:
definition: uppercase;receive;send;source
stream:
bindings:
uppercase-in-0:
destination: test-out
error-handler-definition: errorHandler
uppercase-out-0:
destination: test-in
sink-in-0:
destination: test-in
source-out-0:
destination: test-out
Error Handling
We can attach an error handler to a function by defining a specific Consumer bean with ErrorMessage as input type:
@Bean
public Consumer<ErrorMessage> errorHandler() {
return error -> {
System.out.println(error.getPayload());
System.out.println(error.getOriginalMessage());
};
}
The error handler can then be associated with the function like in the following example:
stream:
bindings:
uppercase-in-0:
destination: test-out
error-handler-definition: errorHandler
Message-Driven Services Triggered by REST Calls
If we need to produce messages by some external REST call, we have to use a StreamBridge service. We can inject this service into a controller and use its send method, passing to it a binding and a payload:
@RestController
@RequestMapping("/booksmanagement")
public class BooksManagementController {
@Autowired
private StreamBridge streamBridge;
@GetMapping(value = "/sendBookInfo")
public void sendBookInfo() {
BookInfo bookInfo = new BookInfo();
bookInfo.setId(1L);
Book book = new Book();
book.setTitle("The Naked Sun");
book.setGenre("Sci-fi");
Author author = new Author();
author.setName("Isaac Asimov");
author.setBiographyInfo("Biography Info");
System.out.println("Sending " + bookInfo);
streamBridge.send("sendBookInfo-out-0", bookInfo);
}
In the application.xml we then define a function with the same name as the controller method and associate it with the same binding we passed as input to the send method:
cloud:
function:
definition: sendBookInfo
stream:
bindings:
sendBookInfo-out-0:
destination: bookInfo-out
Grouping
In micro-service applications, we normally run multiple instances for each service. If the micro-service acts as a consumer for a specific topic, we normally do not want all the instances consuming the same message. We can avoid this by using the grouping functionality. Defining a group for a binding definition allows only one instance of the group at a time to consume a specific message.
For instance, we can implement a sample by sending messages to a topic using a StreamBridge service:
@GetMapping(value = "/sendBookInfoGrouped")
public void sendBookInfoGrouped() {
BookInfo bookInfo = new BookInfo();
bookInfo.setId(1L);
Book book = new Book();
book.setTitle("The Naked Sun");
book.setGenre("Sci-fi");
Author author = new Author();
author.setName("Isaac Asimov");
author.setBiographyInfo("Biography Info");
System.out.println("Sending " + bookInfo);
streamBridge.send("sendBookInfoGrouped-out-0", bookInfo);
}
We define the binding in the configuration file of the producer side:
cloud:
function:
definition: sendBookInfoGrouped
stream:
bindings:
sendBookInfoGrouped-out-0:
destination: bookInfoGrouped-out
And in the consumer part, we define a group for the binding definition:
cloud:
function:
definition: receiveBookInfoGroup
stream:
bindings:
receiveBookInfoGroup-in-0:
destination: bookInfoGrouped-out
group: bookInfoGroupBooks
Partitioning
A typical requirement in message-driven architectures is to partition the input data so single service instances elaborate single partitions. To partition the data we have to use some attribute of the payload. In the example below we add the id attribute of a BookInfo object as a property of the header, named bookInfoId.
@GetMapping(value = "/sendBookInfoPartitioned")
public void sendBookInfoPartitioned() {
BookInfo bookInfo = new BookInfo();
bookInfo.setId(1L);
Book book = new Book();
book.setTitle("The Naked Sun");
book.setGenre("Sci-fi");
Author author = new Author();
author.setName("Isaac Asimov");
author.setBiographyInfo("Biography Info");
System.out.println("Sending " + bookInfo);
streamBridge.send("sendBookInfoPartitioned-out-0", MessageBuilder.withPayload(bookInfo)
.setHeader("bookInfoId", bookInfo.getId())
.build());
}
In the binding configuration, we need two additional properties:
- partitionKeyExpression: headers[‘bookInfoId’]
- partitionCount: 4
As in the following application.xml fragment:
cloud:
function:
definition: sendBookInfoPartitioned
stream:
bindings:
sendBookInfoPartitioned-out-0:
destination: bookInfoPartitioned-out
producer:
partitionKeyExpression: headers['bookInfoId']
partitionCount: 4
The partitionKeyExpression property defines in general an expression that serves the purpose of partitioning the data. It could be any expression, and in this case, it is the bookInfoId attribute retrieved from the header. The partitionCount property defines how many partitions are used. The consumer part will use this number to decide which instance will consume a particular partition.
By default, Spring Cloud Stream will calculate the hash of the partitionKeyExpression property and the module of it and partitionCount: key.hash()%partitionCount. In our example, this will be a number from 0 to 3, and on the consumer side we can set which partition each instance will elaborate by a binding configuration like this:
cloud:
function:
definition: receiveBookInfoPartitioned
stream:
bindings:
receiveBookInfoPartitioned-in-0:
destination: bookInfoPartitioned-out
group: bookInfoGroupBooks
consumer:
partitioned: true
instance-index: 0
instance-count: 4
We define the consumer as partitioned setting the partitioned parameter to true, and assign a number from 0 to 3 to the instance-index parameter. The instance-count parameter defines the number of instances working on the partitioned data.
Conclusion
The message-driven style of communication has particular importance in micro-service systems. It allows for asynchronous interaction and loose coupling of the services. You can find the source code of the example of this article on GitHub.