Hi devs,
RabbitMQ is one of the most popular message brokers, enabling communication between services in a distributed system. It supports various messaging patterns, such as work queues, publish/subscribe, and RPC. In this post, I’ll demonstrate how to set up RabbitMQ and use it to enable communication between three microservices: Order Service, Inventory Service, and Notification 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 an
OrderPlacedmessage 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 at
http://localhost:15672(default username/password:guest/guest).
- Install the
RabbitMQ.Clientpackage in your .NET projects:
dotnet add package RabbitMQ.Client
Step-by-Step Implementation
Step 1: Define a Shared Order Model
public class Order
{
public int Id { get; set; }
public string ProductName { get; set; }
public int Quantity { get; set; }
public decimal TotalPrice { get; set; }
}
This model will represent the order data shared between services.
Step 2: Create the Order Service (Producer)
The Order Service publishes messages to the RabbitMQ exchange.
using RabbitMQ.Client;
using System.Text;
using System.Text.Json;
public class OrderService
{
private readonly IModel _channel;
public OrderService()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
var connection = factory.CreateConnection();
_channel = connection.CreateModel();
// Declare an exchange and a queue
_channel.ExchangeDeclare(exchange: "orders_exchange", type: ExchangeType.Fanout);
}
public void PlaceOrder(Order order)
{
var message = JsonSerializer.Serialize(order);
var body = Encoding.UTF8.GetBytes(message);
_channel.BasicPublish(exchange: "orders_exchange", routingKey: "", basicProperties: null, body: body);
Console.WriteLine($"Order placed: {order.ProductName}");
}
}
// Usage
var orderService = new OrderService();
var order = new Order { Id = 1, ProductName = "Laptop", Quantity = 1, TotalPrice = 1500.00m };
orderService.PlaceOrder(order);
Here, the Order Service publishes an OrderPlaced event to the orders_exchange.
Step 3: Create the Inventory Service (Consumer)
The Inventory Service listens to the orders_exchange and updates stock.
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;
public class InventoryService
{
public void Start()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
// Declare the exchange and queue
channel.ExchangeDeclare(exchange: "orders_exchange", type: ExchangeType.Fanout);
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: queueName, exchange: "orders_exchange", routingKey: "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
var order = 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...");
}
}
// Usage
var inventoryService = new InventoryService();
inventoryService.Start();
Step 4: Create the Notification Service (Consumer)
The Notification Service listens to the orders_exchange and sends notifications.
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;
public class NotificationService
{
public void Start()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
// Declare the exchange and queue
channel.ExchangeDeclare(exchange: "orders_exchange", type: ExchangeType.Fanout);
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: queueName, exchange: "orders_exchange", routingKey: "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
var order = 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...");
}
}
// Usage
var notificationService = new NotificationService();
notificationService.Start();
How It All Works
-
Order Service publishes an
OrderPlacedevent to the RabbitMQorders_exchange. - RabbitMQ routes the message to all bound queues.
- Inventory Service and Notification Service consume the message and process it independently.
Testing the System
- Run the Inventory Service and Notification Service.
- Place an order using the Order 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 the Fanout Exchange pattern, we ensured that multiple services could respond to the same event independently.
Keep coding!
Top comments (0)