Posted onNov 24, 2024

RabbitMQ: Using It with Microservices

#eventdriven #cleancode #microservices #architecture

Hi devs,

RabbitMQ is one of the most popularmessage brokers, enabling communication between services in a distributed system. It supports various messaging patterns, such aswork queues,publish/subscribe, andRPC. In this post, I’ll demonstrate how to set up RabbitMQ and use it to enable communication between three microservices:Order Service,Inventory Service, andNotification Service.

Why RabbitMQ?

RabbitMQ allows microservices to:

Decouple communication: Services don't need to know about each other.
Increase reliability: Messages are persisted until delivered.
Improve scalability: Multiple consumers can process messages concurrently.
Support multiple protocols: AMQP, STOMP, MQTT, etc.

Architecture Overview

In this example:

Order Service publishes anOrderPlaced message to a RabbitMQ exchange.
Inventory Service subscribes to the exchange to update stock.
Notification Service subscribes to the exchange to send order confirmation notifications.

Prerequisites

Install RabbitMQ:

Using Docker:

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

Access the RabbitMQ management UI athttp://localhost:15672 (default username/password:guest/guest).

Install theRabbitMQ.Client package in your .NET projects:

   dotnet add package RabbitMQ.Client

Step-by-Step Implementation

Step 1: Define a Shared Order Model

publicclassOrder{publicintId{get;set;}publicstringProductName{get;set;}publicintQuantity{get;set;}publicdecimalTotalPrice{get;set;}}

This model will represent the order data shared between services.

Step 2: Create the Order Service (Producer)

TheOrder Service publishes messages to the RabbitMQ exchange.

usingRabbitMQ.Client;usingSystem.Text;usingSystem.Text.Json;publicclassOrderService{privatereadonlyIModel_channel;publicOrderService(){varfactory=newConnectionFactory(){HostName="localhost"};varconnection=factory.CreateConnection();_channel=connection.CreateModel();// Declare an exchange and a queue_channel.ExchangeDeclare(exchange:"orders_exchange",type:ExchangeType.Fanout);}publicvoidPlaceOrder(Orderorder){varmessage=JsonSerializer.Serialize(order);varbody=Encoding.UTF8.GetBytes(message);_channel.BasicPublish(exchange:"orders_exchange",routingKey:"",basicProperties:null,body:body);Console.WriteLine($"Order placed:{order.ProductName}");}}// UsagevarorderService=newOrderService();varorder=newOrder{Id=1,ProductName="Laptop",Quantity=1,TotalPrice=1500.00m};orderService.PlaceOrder(order);

Here, theOrder Service publishes anOrderPlaced event to theorders_exchange.

Step 3: Create the Inventory Service (Consumer)

TheInventory Service listens to theorders_exchange and updates stock.

usingRabbitMQ.Client;usingRabbitMQ.Client.Events;usingSystem.Text;usingSystem.Text.Json;publicclassInventoryService{publicvoidStart(){varfactory=newConnectionFactory(){HostName="localhost"};varconnection=factory.CreateConnection();varchannel=connection.CreateModel();// Declare the exchange and queuechannel.ExchangeDeclare(exchange:"orders_exchange",type:ExchangeType.Fanout);varqueueName=channel.QueueDeclare().QueueName;channel.QueueBind(queue:queueName,exchange:"orders_exchange",routingKey:"");varconsumer=newEventingBasicConsumer(channel);consumer.Received+=(model,ea)=>{varbody=ea.Body.ToArray();varmessage=Encoding.UTF8.GetString(body);varorder=JsonSerializer.Deserialize<Order>(message);Console.WriteLine($"Inventory updated for Product:{order.ProductName}, Quantity:{order.Quantity}");};channel.BasicConsume(queue:queueName,autoAck:true,consumer:consumer);Console.WriteLine("Inventory Service is running...");}}// UsagevarinventoryService=newInventoryService();inventoryService.Start();

Step 4: Create the Notification Service (Consumer)

TheNotification Service listens to theorders_exchange and sends notifications.

usingRabbitMQ.Client;usingRabbitMQ.Client.Events;usingSystem.Text;usingSystem.Text.Json;publicclassNotificationService{publicvoidStart(){varfactory=newConnectionFactory(){HostName="localhost"};varconnection=factory.CreateConnection();varchannel=connection.CreateModel();// Declare the exchange and queuechannel.ExchangeDeclare(exchange:"orders_exchange",type:ExchangeType.Fanout);varqueueName=channel.QueueDeclare().QueueName;channel.QueueBind(queue:queueName,exchange:"orders_exchange",routingKey:"");varconsumer=newEventingBasicConsumer(channel);consumer.Received+=(model,ea)=>{varbody=ea.Body.ToArray();varmessage=Encoding.UTF8.GetString(body);varorder=JsonSerializer.Deserialize<Order>(message);Console.WriteLine($"Notification sent for Order ID:{order.Id}, Product:{order.ProductName}");};channel.BasicConsume(queue:queueName,autoAck:true,consumer:consumer);Console.WriteLine("Notification Service is running...");}}// UsagevarnotificationService=newNotificationService();notificationService.Start();

How It All Works

Order Service publishes anOrderPlaced event to the RabbitMQorders_exchange.
RabbitMQ routes the message to all bound queues.
Inventory Service andNotification Service consume the message and process it independently.

Testing the System

Run theInventory Service andNotification Service.
Place an order using theOrder Service.
Observe the logs for the inventory update and notification.

Benefits of Using RabbitMQ in Microservices

Decoupled Communication: Producers and consumers don’t need direct knowledge of each other.
Scalable: Add more consumers to process messages faster.
Reliable Delivery: Messages are persisted and retried if delivery fails.
Flexible Messaging Patterns: Work queues, publish/subscribe, RPC, etc.

Conclusion

RabbitMQ is a powerful message broker that simplifies communication in microservices. By using theFanout Exchange pattern, we ensured that multiple services could respond to the same event independently.

Keep coding!