Event driven microservices with spring cloud stream binder rabbit

"Human beings are always in a state of motion, interacting with their environment while sending out and receiving information from the things around them. Typically, these conversations aren’t synchronous, linear, or narrowly defined to a request-response model. It’s something more like a message-driven, where we’re constantly sending and receiving messages. As we receive messages, we react to those messages, while often interrupting the primary task that we’re working on."

-- Spring Microservices in Action

A monolithic application is a tightly coupled big ball of mud. Microservices are loosely coupled highly cohesive in nature self-contained applications. Coupling here refers to the degree to which one application knows or uses-service of another application. Cohesion defines a single well-defined role of an application. Any code dumped inside a monolith application without proper tests can bring the whole system to halt.

• Loose coupling is a desirable state where applications minimize direct reference to each other.
• Tight coupling is an undesirable state where applications are directly referring to each other.
• Highly cohesive in nature means the application is well focused and does what it is supposed to do.
• Low cohesive applications tend to perform many tasks without focusing on a single feature.

When we talk about messaging framework usually called message-broker we deviate away from the traditional way of synchronous request-response style of communication. Applications having direct references once separated now communicate using message-broker. Here is how a traditional messaging infrastructure looks like Messaging Infrastructure consists of 2 styles of communication:

• Point-To-Point channel (https://www.enterpriseintegrationpatterns.com/patterns/messaging/PointToPointChannel.html)
• Publisher-Subscriber Channel (https://www.enterpriseintegrationpatterns.com/patterns/messaging/PublishSubscribeChannel.html)

Example

Our simple application has 2 components: a user component and a birthday component. Whenever a new user is added to the system, we will replicate that user to another service via indirect communication. Here the user service is completely independent of the birthday service. The only thing user service will do is publish a message to the message-broker. These 2 components do not have any direct reference to each other hence if one of them goes down the system is still working with other components working as expected. Birthday service subscribes to the channel in this case userCreated receives the event which contains the user object, for its business processing birthday service transforms that object into BirthdayReminder. The entire idea is whenever a user gets registered into a system, send his information down-stream to process and perform business operations, one of them sending a birthday reminder to a user whose birthday is today or send some promotional message or provide some offers on services to the users who are about to celebrate their birthday.

Install message broker

1docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

visit http://localhost:15672/ and provide guest/guest credentials to access the management console.

Implementation

We are using spring cloud stream with rabbitmq bindings, let's add the following dependency to the gradle.

1implementation 'org.springframework.cloud:spring-cloud-stream-binder-rabbit'

We use StreamBridge to send message to target exchange.

Whenever a new user registers into the system we send his details via message channel as shown below on the producer side.

 1...
 2public static final String USER_CREATED_BINDING_NAME = "userCreated";
 3
 4@PostMapping
 5public Map<String, String> create(@RequestBody User user) {
 6   final UserCreatedEvent userCreatedEvent = UserCreatedEvent.generateForUser(user);
 7   userDatastore.add(userCreatedEvent.getUser());
 8   final boolean sent = streamBridge.send(USER_CREATED_BINDING_NAME, userCreatedEvent);
 9   System.out.printf("Message sent %s%n", sent);
10   return Map.of("status", "success");
11}
12...

On the consumer side, we have to create a bean as shown below and declare binding in application.yml notice that the name of the binding is the same as the bean created.

 1...
 2@Bean
 3public Consumer<UserCreatedEvent> userCreated() {
 4   return userCreatedEvent -> {
 5       logger.info(userCreatedEvent.toString());
 6       BirthdayReminder birthdayReminder = new BirthdayReminder(UUID.randomUUID().toString(), false, userCreatedEvent.getUser());
 7       add(birthdayReminder);
 8   };
 9}
10...

application.yml

 1server:
 2 port: 7003
 3spring:
 4 application:
 5   name: birthday-service
 6 cloud:
 7   stream:
 8     bindings:
 9       userCreated-in-0:
10         destination: userCreated
11         group: birthday-microservice-group
12
13logging:
14 level:
15   org:
16     springframework:
17       amqp: OFF

Once the receiver picks up the message from the channel we convert it and store it locally and filter for upcoming and today's birthday in a scheduled method

 1public List<BirthdayReminder> listTodaysBirthday() {
 2   final LocalDate today = LocalDate.now();
 3   return list(birthdayReminder -> today.isEqual(birthdayReminder.getUser().getDateOfBirth()));
 4}
 5public List<BirthdayReminder> listUpcomingBirthdays() {
 6   final LocalDate today = LocalDate.now();
 7   return list(birthdayReminder -> birthdayReminder.getUser().getDateOfBirth().isAfter(today));
 8}
 9
10public List<BirthdayReminder> list(Predicate<BirthdayReminder> predicate) {
11   return users.entrySet()
12           .stream()
13           .parallel()
14           .map(Map.Entry::getValue)
15           .filter(predicate)
16           .collect(Collectors.toUnmodifiableList());
17}

Output

Source Code

https://gitlab.com/spring-boot-cloud-samples/spring-boot-cloud-stream-binder-rabbitmq