RabbitMQ Message Queue

Overview

RabbitMQ is an open-source message broker that implements AMQP (Advanced Message Queuing Protocol) for asynchronous communication in distributed systems.

1. Installation and Configuration

1.1 Installing with Docker

# Pull RabbitMQ image
docker pull rabbitmq:3-management

# Start container
docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=password \
  rabbitmq:3-management

1.2 Accessing the Management UI

Open your browser and visit http://localhost:15672, then log in with username admin and password password.

2. Core Concepts

2.1 Message Model

Producer -> Exchange -> Queue -> Consumer

2.2 Exchange Types

Type	Description
Direct	Exact routing key match
Fanout	Broadcast to all bound queues
Topic	Pattern-based routing key match
Headers	Match based on message headers

3. Using the .NET Client

3.1 Installing Dependencies

dotnet add package RabbitMQ.Client

3.2 Producer Implementation

using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory() 
{ 
    HostName = "localhost",
    UserName = "admin",
    Password = "password"
};

using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    // Declare queue
    channel.QueueDeclare(queue: "hello",
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    string message = "Hello RabbitMQ!";
    var body = Encoding.UTF8.GetBytes(message);

    // Publish message
    channel.BasicPublish(exchange: "",
                         routingKey: "hello",
                         basicProperties: null,
                         body: body);

    Console.WriteLine($"发送消息: {message}");
}

Console.WriteLine("按任意键退出...");
Console.ReadKey();

3.3 Consumer Implementation

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var factory = new ConnectionFactory() 
{ 
    HostName = "localhost",
    UserName = "admin",
    Password = "password"
};

using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    // Declare queue (same as producer)
    channel.QueueDeclare(queue: "hello",
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        Console.WriteLine($"收到消息: {message}");
    };

    // Consume messages
    channel.BasicConsume(queue: "hello",
                         autoAck: true,
                         consumer: consumer);

    Console.WriteLine("等待消息...");
    Console.ReadKey();
}

4. Advanced Features

4.1 Message Persistence

// Declare durable queue
channel.QueueDeclare(queue: "task_queue",
                     durable: true,  // queue persistence
                     exclusive: false,
                     autoDelete: false,
                     arguments: null);

// Send persistent message
var properties = channel.CreateBasicProperties();
properties.Persistent = true;  // message persistence

channel.BasicPublish(exchange: "",
                     routingKey: "task_queue",
                     basicProperties: properties,
                     body: body);

4.2 Message Acknowledgment

// Disable auto-acknowledgment
channel.BasicConsume(queue: "task_queue",
                     autoAck: false,  // manual acknowledgment
                     consumer: consumer);

consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine($"收到消息: {message}");

    // Simulate processing time
    Thread.Sleep(1000);

    Console.WriteLine("消息处理完成");
    
    // Manually acknowledge message
    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};

4.3 Fair Dispatch

// Receive only one unacknowledged message at a time
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

5. Using Exchanges

5.1 Direct Exchange

// Declare Direct exchange
channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);

// Bind queues to exchange
channel.QueueBind(queue: "queue1", exchange: "direct_logs", routingKey: "error");
channel.QueueBind(queue: "queue2", exchange: "direct_logs", routingKey: "info");
channel.QueueBind(queue: "queue2", exchange: "direct_logs", routingKey: "warning");

// Send message to exchange
channel.BasicPublish(exchange: "direct_logs",
                     routingKey: "error",
                     basicProperties: null,
                     body: body);

5.2 Fanout Exchange

// Declare Fanout exchange
channel.ExchangeDeclare(exchange: "fanout_logs", type: ExchangeType.Fanout);

// Bind queues to exchange (routing key is ignored)
channel.QueueBind(queue: "queue1", exchange: "fanout_logs", routingKey: "");
channel.QueueBind(queue: "queue2", exchange: "fanout_logs", routingKey: "");

// Send message (routing key is ignored)
channel.BasicPublish(exchange: "fanout_logs",
                     routingKey: "",
                     basicProperties: null,
                     body: body);

5.3 Topic Exchange

// Declare Topic exchange
channel.ExchangeDeclare(exchange: "topic_logs", type: ExchangeType.Topic);

// Bind queues (using wildcards)
channel.QueueBind(queue: "queue1", exchange: "topic_logs", routingKey: "*.orange.*");
channel.QueueBind(queue: "queue2", exchange: "topic_logs", routingKey: "*.*.rabbit");
channel.QueueBind(queue: "queue2", exchange: "topic_logs", routingKey: "lazy.#");

// Send message
channel.BasicPublish(exchange: "topic_logs",
                     routingKey: "quick.orange.rabbit",
                     basicProperties: null,
                     body: body);

6. Dead Letter Queue

6.1 Creating Dead Letter Exchange and Queue

// Declare dead letter exchange
channel.ExchangeDeclare(exchange: "dlx_exchange", type: ExchangeType.Direct);

// Declare dead letter queue
channel.QueueDeclare(queue: "dlx_queue", durable: true, exclusive: false, autoDelete: false);

// Bind dead letter queue to dead letter exchange
channel.QueueBind(queue: "dlx_queue", exchange: "dlx_exchange", routingKey: "#");

// Declare normal queue with dead letter exchange
var args = new Dictionary<string, object>
{
    { "x-dead-letter-exchange", "dlx_exchange" },
    { "x-dead-letter-routing-key", "#" }
};
channel.QueueDeclare(queue: "normal_queue", durable: true, exclusive: false, autoDelete: false, arguments: args);

6.2 Message Expiration

// Set message expiration time (milliseconds)
var properties = channel.CreateBasicProperties();
properties.Expiration = "60000"; // expires in 60 seconds

channel.BasicPublish(exchange: "",
                     routingKey: "normal_queue",
                     basicProperties: properties,
                     body: body);

7. Practical Recommendations

7.1 Best Practices

1. Queue naming conventions: Use meaningful queue names
2. Message persistence: Enable persistence in production environments
3. Message acknowledgment: Use manual acknowledgment to prevent message loss
4. Message expiration: Set reasonable message expiration times
5. Monitoring and alerts: Monitor queue length and message backlog

7.2 Common Issues

Issue	Solution
Message loss	Enable persistence, use manual acknowledgment
Duplicate messages	Message deduplication, idempotent design
Queue backlog	Add consumers, optimize consumption speed
Network issues	Set reasonable timeouts and retry mechanisms

Summary

RabbitMQ is a powerful message queue system that supports multiple messaging patterns and advanced features. Proper usage can improve system asynchronous processing capabilities and reliability.

---

Author: Lei Tao
Date: April 15, 2024