Spring Boot Microservice Use Cases: Messaging with RabbitMQ

In today’s fast-paced world of software, making applications talk to each other efficiently is crucial. Imagine RabbitMQ as a smart postman for your software — it takes messages from one place and delivers them to another, making sure they end up in the right hands. But it doesn’t stop there; RabbitMQ’s superpower is in its different message-routing strategies, known as exchange types: Direct Exchange, Fanout Exchange, and Topic Exchange. We’ll use real-world examples to help you understand how these tricks work, making you the magician of your software communication.

Fundamental Components of RabbitMQ:

1. Producer: The producer is responsible for creating and sending messages to RabbitMQ. It generates messages and specifies which exchange to send them to, along with a routing key. Messages can be any data that needs to be communicated between applications.
2. Exchange: The exchange is a routing mechanism that receives messages from producers and routes them to the appropriate queues. RabbitMQ offers different types of exchanges, including direct, topic, and fanout, each with its routing strategy.
3. Queue: Queues are message buffers that store messages until they are consumed by consumers. Multiple queues can be bound to an exchange, and each queue is responsible for specific message processing.
4. Binding: A binding is a link that connects an exchange to one or more queues. It defines the rules for how messages should be routed from the exchange to the queues based on routing keys and patterns.
5. Consumer: Consumers are applications or services that receive and process messages from queues. They subscribe to specific queues and consume messages as they become available. Consumers perform tasks such as order processing, data analysis, or any action based on the message content.

Principal Messaging Workflow:

• The producer publishes a message to exchange.
• After receiving the message, the exchange is responsible for forwarding it. The exchange routes the message to queues, exchanges bound to it.
• Queue receives the message and keeps until the consumer consumes it.
• Lastly, the consumer handles the message.

Direct Exchange:

The exchange forwards the message to a queue based on a routing key. The routing key is specified by the producer when publishing a message. Messages with matching routing keys are routed to the corresponding queues.

SCENARIO

• order_create_queue is bound with binding key order-create and order_create_log_queue is bound with order-create-log routing key.
• When a new message with routing key order-create arrives at the direct exchange, the exchange routes it to the queue with the appropriate routing key, in this case to queue order_create_queue.

Use Cases:

1. Log Routing: When logging from multiple components of an application, direct exchanges can be used to route logs to different destinations based on severity levels. For example, you can have queues for “info,” “warning,” and “error” logs, each with its routing key.
2. Order Processing: In an e-commerce system, you can use a direct exchange to process orders. Different queues can be created for order processing stages such as payment, inventory management, and shipping. Each order is assigned a routing key based on its status, and the exchange routes orders to the appropriate queues for processing.

Direct Exchange: Order processing in an e-commerce

Create a Configuration Class:

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class OrderProcessingConfig {

    public static final String ORDER_DIRECT_EXCHANGE = "order-direct-exchange";
    public static final String PAYMENT_QUEUE = "payment-queue";
    public static final String INVENTORY_QUEUE = "inventory-queue";
    public static final String SHIPPING_QUEUE = "shipping-queue";

    // Routing keys for different order processing stages
    public static final String ROUTING_KEY_PAYMENT = "payment";
    public static final String ROUTING_KEY_INVENTORY = "inventory";
    public static final String ROUTING_KEY_SHIPPING = "shipping";

    @Bean
    public DirectExchange orderDirectExchange() {
        return new DirectExchange(ORDER_DIRECT_EXCHANGE);
    }

    @Bean
    public Queue paymentQueue() {
        return new Queue(PAYMENT_QUEUE);
    }

    @Bean
    public Queue inventoryQueue() {
        return new Queue(INVENTORY_QUEUE);
    }

    @Bean
    public Queue shippingQueue() {
        return new Queue(SHIPPING_QUEUE);
    }

    @Bean
    public Binding paymentBinding() {
        return BindingBuilder.bind(paymentQueue()).to(orderDirectExchange()).with(ROUTING_KEY_PAYMENT);
    }

    @Bean
    public Binding inventoryBinding() {
        return BindingBuilder.bind(inventoryQueue()).to(orderDirectExchange()).with(ROUTING_KEY_INVENTORY);
    }

    @Bean
    public Binding shippingBinding() {
        return BindingBuilder.bind(shippingQueue()).to(orderDirectExchange()).with(ROUTING_KEY_SHIPPING);
    }
}

Order Producer:

Create a producer to send tasks to the Direct Exchange based on routing keys:

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class OrderProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void processOrderPayment(String orderId) {
        // Simulate processing payment and send order to the payment queue
        rabbitTemplate.convertAndSend(OrderProcessingConfig.ORDER_DIRECT_EXCHANGE, OrderProcessingConfig.ROUTING_KEY_PAYMENT, orderId);
    }

    public void processOrderInventory(String orderId) {
        // Simulate managing inventory and send order to the inventory queue
        rabbitTemplate.convertAndSend(OrderProcessingConfig.ORDER_DIRECT_EXCHANGE, OrderProcessingConfig.ROUTING_KEY_INVENTORY, orderId);
    }

    public void processOrderShipping(String orderId) {
        // Simulate shipping and send order to the shipping queue
        rabbitTemplate.convertAndSend(OrderProcessingConfig.ORDER_DIRECT_EXCHANGE, OrderProcessingConfig.ROUTING_KEY_SHIPPING, orderId);
    }
}

Order Processing Consumers:

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class PaymentProcessor {

    @RabbitListener(queues = OrderProcessingConfig.PAYMENT_QUEUE)
    public void processPayment(String orderId) {
        // Process payment logic for the order
        System.out.println("Payment processed for order: " + orderId);
    }
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class InventoryManager {

    @RabbitListener(queues = OrderProcessingConfig.INVENTORY_QUEUE)
    public void manageInventory(String orderId) {
        // Manage inventory logic for the order
        System.out.println("Inventory managed for order: " + orderId);
    }
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class ShippingHandler {

    @RabbitListener(queues = OrderProcessingConfig.SHIPPING_QUEUE)
    public void handleShipping(String orderId) {
        // Handle shipping logic for the order
        System.out.println("Order shipped: " + orderId);
    }
}

Processing Orders:

@Autowired
private OrderProducer orderProducer;

// Simulate order processing stages
orderProducer.processOrderPayment("12345");
orderProducer.processOrderInventory("12345");
orderProducer.processOrderShipping("12345");

Direct Exchange- Microservice Usages:

Service-to-Service Communication:

Direct exchanges are commonly used for point-to-point communication between microservices. Each microservice can have its own queue bound to a direct exchange, and messages are routed based on a specific routing key. This ensures that messages are delivered directly to the intended microservice, making it a suitable choice for synchronous communication.

• The producer microservice (MessageProducerService) sends a message to the RabbitMQ exchange named "microservice-direct-exchange" using a specific routing key ("microservice-routing-key").
• The consumer microservice (MessageConsumerService) listens for messages on the "queue-for-microservice" queue, which is bound to the same exchange and routing key.

Event Notification Communication:

publish events from one microservice and have another microservice subscribe to those events using a direct exchange. In this example, we’ll create a simple “Order” microservice that publishes order creation events, and a “Notification” microservice that subscribes to these events.

Order Microservice:

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderService {

    private final AmqpTemplate amqpTemplate;

    @Autowired
    public OrderService(AmqpTemplate amqpTemplate) {
        this.amqpTemplate = amqpTemplate;
    }

    public void createOrder(Order order) {
        // Business logic to create the order
        // ...

        // Publish an order creation event with a routing key "order.created"
        amqpTemplate.convertAndSend("orders-direct-exchange", "order.created", order);
    }
}

Notification Microservice:

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class NotificationService {

    @RabbitListener(queues = "notification-queue")
    public void receiveOrderCreatedEvent(Order order) {
        // Handle the order creation event
        // Send a notification, log, or perform any desired action
        System.out.println("Received order created event: " + order);
    }
}

RabbitMQ Configuration: In both microservices, you’ll need to configure the RabbitMQ connection and declare the “orders-direct-exchange” and “notification-queue” as shown in the previous responses.

Fanout Exchange:

A Fanout Exchange in RabbitMQ is used when you want to broadcast a message to multiple queues, regardless of the routing key or the message content. It’s often used for scenarios where you need to send the same message to multiple consumers or components.

SCENARIO 1

• Exchange: sport_news
• Queue A: Mobile client queue A
• Binding: Binding between the exchange (sport_news) and Queue A (Mobile client queue A)

Use Cases:

1. Real-time Dashboard Updates: In a monitoring system or dashboard, you might want to send updates to multiple clients simultaneously. Each client could have its own queue, and a Fanout Exchange would broadcast updates to all connected clients.
2. Logs and Notifications: When you want to log messages and send notifications to multiple destinations, such as email, SMS, and a log storage system, a Fanout Exchange can be useful.

Fan-Out Exchange: Building Notification and logging system

Create a Configuration Class:

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class NotificationConfig {

    public static final String NOTIFICATION_FANOUT_EXCHANGE = "notification-fanout-exchange";

    @Bean
    public FanoutExchange notificationFanoutExchange() {
        return new FanoutExchange(NOTIFICATION_FANOUT_EXCHANGE);
    }

    @Bean
    public Queue emailQueue() {
        return new Queue("email-queue");
    }

    @Bean
    public Queue smsQueue() {
        return new Queue("sms-queue");
    }

    @Bean
    public Queue logStorageQueue() {
        return new Queue("log-storage-queue");
    }

    @Bean
    public Binding emailBinding() {
        return BindingBuilder.bind(emailQueue()).to(notificationFanoutExchange());
    }

    @Bean
    public Binding smsBinding() {
        return BindingBuilder.bind(smsQueue()).to(notificationFanoutExchange());
    }

    @Bean
    public Binding logStorageBinding() {
        return BindingBuilder.bind(logStorageQueue()).to(notificationFanoutExchange());
    }
}

Notification Producer: While sending message just mention the exchange name

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class NotificationProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private FanoutExchange notificationFanoutExchange;

    public void sendEmailNotification(String message) {
        rabbitTemplate.convertAndSend(notificationFanoutExchange.getName(), "", "Email: " + message);
    }

    public void sendSMSNotification(String message) {
        rabbitTemplate.convertAndSend(notificationFanoutExchange.getName(), "", "SMS: " + message);
    }

    public void sendLogToStorage(String logMessage) {
        rabbitTemplate.convertAndSend(notificationFanoutExchange.getName(), "", "Log: " + logMessage);
    }
}

Notification Consumers: Email, SMS, Log Storage Queue

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class EmailNotificationConsumer {

    @RabbitListener(queues = "email-queue")
    public void sendEmailNotification(String message) {
        // Send email notification logic
        System.out.println("Email Notification: " + message);
    }
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SMSNotificationConsumer {

    @RabbitListener(queues = "sms-queue")
    public void sendSMSNotification(String message) {
        // Send SMS notification logic
        System.out.println("SMS Notification: " + message);
    }
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class LogStorageConsumer {

    @RabbitListener(queues = "log-storage-queue")
    public void storeLog(String message) {
        // Store log message to a log storage system
        System.out.println("Log Storage: " + message);
    }
}

Sending Notifications and Logs: In your application code, you can use the NotificationProducer to send notifications and log messages to all consumers:

@Autowired
private NotificationProducer notificationProducer;

// Send email notification
notificationProducer.sendEmailNotification("New order placed.");

// Send SMS notification
notificationProducer.sendSMSNotification("Payment successful.");

// Store log message
notificationProducer.sendLogToStorage("Application error occurred.");

Create a Fanout Exchange named “notification-fanout-exchange” and three queues: “email-queue,” “sms-queue,” and “log-storage-queue,” all bound to the exchange. Notifications and log messages sent to the exchange will be broadcasted to all queues, allowing email, SMS, and log storage systems to receive and process the messages simultaneously.

Fanout Exchange: Broadcast Notifications in Microservice

Fanout exchanges are often used to broadcast notifications, events, or updates to multiple microservices simultaneously. For example, when a user makes a change to their profile, a User microservice can publish a profile update event to a fanout exchange, and several other microservices (e.g., Notification, Dashboard, Email) can subscribe to this exchange to receive and process the event independently.

Topic Exchange:

The exchange routes the message to bounded queues based on a pattern defined on the exchange and the routing keys attached to the queues. It allows for more flexible routing based on wildcard patterns.

The power of the Topic Exchange comes from the use of two wildcard characters:

• * (star): Matches a single word or segment in the routing key.
• # (hash): Matches zero or more words or segments in the routing key.

This allows you to create complex patterns for routing. For example:

• Routing key binding with “sports.#” would match publisher’s message routing key “sports.football” and “sports.tennis”. (Like Fanout)
• Routing key binding with “user.*.login” would match publisher’s message routing key “user.john.login” and “user.mary.login”.

Topic Exchange: E-commerce Example

1. A Topic Exchange is set up.
2. There are three queues:

• customer_orders_queue wants to receive all customer order logs, is bound with the pattern order.logs.customer.#, which matches any producer message routing keys starting with "order.logs.customer ".
• all_order_logs_queue wants to receive all order logs, bound with the pattern order.logs.#, which matches any producer message routing keys starting with "order.logs ".
• electronics_order_logs_queue wants to receive all electronics order logs, is bound with the pattern order.logs.*.electronics, which matches producer message routing keys where the fourth word is "electronics."

3. When messages are published to the Topic Exchange with various routing keys, they are routed to the appropriate queues based on the matching patterns. For example, if a message with the routing key order.logs.customer.electronics is published, it will be routed to both the customer_orders_queue and the electronics_order_logs_queue because it matches the patterns defined for both queues.

Configuration:

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    // Define a TopicExchange named "order-logs-exchange"
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("order-logs-exchange");
    }

    // Define a Queue named "customer_orders_queue"
    @Bean
    public Queue customerOrdersQueue() {
        return new Queue("customer_orders_queue");
    }

    // Define a Queue named "all_order_logs_queue"
    @Bean
    public Queue allOrderLogsQueue() {
        return new Queue("all_order_logs_queue");
    }

    // Define a Queue named "electronics_order_logs_queue"
    @Bean
    public Queue electronicsOrderLogsQueue() {
        return new Queue("electronics_order_logs_queue");
    }

    // Create a binding between "customer_orders_queue" and the exchange with the routing key pattern "order.logs.customer.#"
    @Bean
    public Binding binding1(TopicExchange topicExchange, Queue customerOrdersQueue) {
        return BindingBuilder.bind(customerOrdersQueue)
                .to(topicExchange)
                .with("order.logs.customer.#");
    }

    // Create a binding between "all_order_logs_queue" and the exchange with the routing key pattern "order.logs.#"
    @Bean
    public Binding binding2(TopicExchange topicExchange, Queue allOrderLogsQueue) {
        return BindingBuilder.bind(allOrderLogsQueue)
                .to(topicExchange)
                .with("order.logs.#");
    }

    // Create a binding between "electronics_order_logs_queue" and the exchange with the routing key pattern "order.logs.*.electronics"
    @Bean
    public Binding binding3(TopicExchange topicExchange, Queue electronicsOrderLogsQueue) {
        return BindingBuilder.bind(electronicsOrderLogsQueue)
                .to(topicExchange)
                .with("order.logs.*.electronics");
    }
}

Publish Message:

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MessagePublisher {

    private final AmqpTemplate amqpTemplate;

    @Autowired
    public MessagePublisher(AmqpTemplate amqpTemplate) {
        this.amqpTemplate = amqpTemplate;
    }

    public void publishCustomerOrderLog() {
        String routingKey = "order.logs.customer.electronics";
        String message = "Customer order log for electronics";
        amqpTemplate.convertAndSend("order-logs-exchange", routingKey, message);
    }

    // Add similar methods to publish messages with other routing keys
}

Consumer:

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

@Service
public class CustomerOrderLogConsumer {

    @RabbitListener(queues = "customer_orders_queue")
    public void processCustomerOrderLog(String message) {
        System.out.println("Received in customer_orders_queue: " + message);
    }
}

Headers Exchange:

In this case, the message header attributes are used, instead of the routing key, to bind an exchange to one or more queues. It provides complex matching based on header attributes.

— — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — -

Exchange Properties:

When declaring an exchange, you need to specify several properties, including:

• Name: The name of the exchange, which is used to identify it.
• Durability: If enabled (set to true), the broker will not remove the exchange if it goes offline or in case of a broker restart.
• In RabbitMQ, you can declare queues and messages as durable, which means they will survive broker restarts.
• To ensure durability, messages are written to disk, which can impact performance but guarantees message persistence.

@Bean
public Exchange myExchange() {
    return ExchangeBuilder.directExchange("my-direct-exchange")
            .durable(true)
            .build();
}

• Auto-Delete: If enabled, the broker deletes the queue when the last consumer unsubscribes (i.e., it's not connected to any queues). If set to false, the exchange will persist even if it's not bound to any queues.

@Bean
public Exchange myExchange() {
    return ExchangeBuilder.directExchange("my-direct-exchange")
            .durable(true)
            .autoDelete(true)
            .build();
}

Use Cases

Task Queues:

• Image Processing: In an e-commerce platform, when users upload product images for their listings, RabbitMQ can be used to distribute image processing tasks to multiple worker nodes. Each node can process images, generate thumbnails, and optimize them for web display.
• Email Queue: In a customer support system, RabbitMQ can manage an email queue where incoming support requests are distributed to available customer service agents for prompt response.

Workflows and Coordination:

• Order Processing Workflow: In an e-commerce platform, RabbitMQ can orchestrate the order processing workflow. When a customer places an order, RabbitMQ can ensure that the order goes through sequential steps, such as order validation, payment processing, inventory management, and shipping, in a coordinated manner.
• Document Approval Workflow: In a document management system, RabbitMQ can be used to manage document approval workflows. Documents can be passed from one reviewer to the next in a predefined sequence until final approval is reached.

RPC (Remote Procedure Call):

• Payment Gateway: In a payment processing system, RabbitMQ can facilitate synchronous communication between a web application and a payment gateway. When a user initiates a payment, the web application can use RPC over RabbitMQ to interact with the payment gateway to authorize the transaction and receive a response.
• Geolocation Service: An application that requires geolocation data can use RabbitMQ for RPC to query a geolocation service for latitude and longitude coordinates based on user IP addresses.

Highly Reliable Messaging:

• Financial Transactions: In a financial trading platform, RabbitMQ can be used to ensure the reliable delivery of trade execution messages. Acknowledgments and retries can be implemented to guarantee that trade orders are not lost.
• Healthcare Alerts: In a healthcare monitoring system, RabbitMQ can be employed to transmit critical patient alerts to healthcare providers. It is essential that such alerts are reliably delivered to the right recipients.

Low Latency Requirements:

• Real-time Notifications: In a chat application, RabbitMQ can be used for delivering real-time chat messages with lower latency compared to some other messaging systems. Users can receive messages from their contacts with minimal delay.
• Live Dashboard Updates: In a monitoring and analytics dashboard, RabbitMQ can deliver live updates to the dashboard components, ensuring that data changes are reflected quickly.

In our journey through RabbitMQ, we’ve uncovered a world of messaging magic. From Direct Exchange’s precise deliveries to Fanout Exchange’s broadcasts and Topic Exchange’s wildcard wizardry, RabbitMQ offers a versatile toolkit for crafting effective communication in your software systems.

But RabbitMQ isn’t just about tricks; it’s about empowering your applications to communicate seamlessly and reliably. We’ve explored crucial concepts like concurrency for efficient processing, message acknowledgment for reliability, error handling for graceful recovery, and message persistence for safeguarding data.