RabbitMQ五種消息模型詳解

一、rabbitmq五種消息模型簡介

RabbitMQ是一個開源的消息隊列中間件，支持多種消息傳輸模式，如發布/訂閱、路由、工作隊列、RPC等。在消息傳輸過程中，RabbitMQ將消息存儲在隊列中並轉發給接收方。

RabbitMQ的五種常用消息模型包括：簡單隊列模型、工作隊列模型、發布/訂閱模型、路由模型、主題模型。

二、rabbitmq五種消息模型應用場景

1、簡單隊列模型

應用場景：適用於只有一個消費者的場景，通常用於解耦和最終目標是將消息發布到集體中。

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queueName, false, false, false, null);
    string message = "hello rabbitmq";
    var body = Encoding.UTF8.GetBytes(message);
 
    channel.BasicPublish("", queueName, null, body);
    Console.WriteLine("Send Message：{0}", message);
}

<!-- rabbitmq 消費者 -->
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
    channel.BasicAck(ea.DeliveryTag, false);
};
channel.BasicConsume(queueName, false, consumer);

2、工作隊列模型

應用場景：適用於多個消費者的場景，任務可以平均地在多個消費者之間分配執行。

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queueName, true, false, false, null);
    string message = "hello workQueue";
    var body = Encoding.UTF8.GetBytes(message);
    var properties = channel.CreateBasicProperties();
    properties.Persistent = true;
    channel.BasicPublish("", queueName, properties, body);
    Console.WriteLine("Send Message：{0}", message);
}

<!-- rabbitmq 消費者1 -->
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
    channel.BasicAck(ea.DeliveryTag, false);
};
channel.BasicConsume(queueName, false, consumer);

發送1000條消息，使用兩個消費者消費：

3、發布/訂閱模型

應用場景：適用於多個消費者需要接收同一個消息的場景。

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout);
    string message = "hello fanout";
    var body = Encoding.UTF8.GetBytes(message);
    channel.BasicPublish(exchangeName, "", null, body);
 
    Console.WriteLine("Send Message：{0}", message);
}

<!-- rabbitmq 消費者1 -->
channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout);
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queueName, exchangeName, "");
 
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
};
channel.BasicConsume(queueName, true, consumer);

啟動兩個消費者監聽同一個隊列：

4、路由模型

應用場景：適用於多個消費者根據不同的條件接收不同的消息。

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
    string message = "hello direct";
    var routingKey = "error";
    var body = Encoding.UTF8.GetBytes(message);
    channel.BasicPublish(exchangeName, routingKey, null, body);
 
    Console.WriteLine("Send Message：{0}", message);
}

<!-- rabbitmq 消費者1 -->
var queueName = channel.QueueDeclare().QueueName;
var routingKey = "error";
channel.QueueBind(queueName, exchangeName, routingKey);
 
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
};
channel.BasicConsume(queueName, true, consumer);

5、主題模型

應用場景：適用於多個消費者通過實現通配符功能接收指定消息。

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);
    string message = "hello topic1";
    var routingKey = "usa.news";
    var body = Encoding.UTF8.GetBytes(message);
    channel.BasicPublish(exchangeName, routingKey, null, body);
 
    Console.WriteLine("Send Message：{0}", message);
}

<!-- rabbitmq 消費者1 -->
var queueName = channel.QueueDeclare().QueueName;
var routingKey = "usa.#";
channel.QueueBind(queueName, exchangeName, routingKey);
 
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
};
channel.BasicConsume(queueName, true, consumer);

三、rabbitmq五種消息模型常用

1、c#

RabbitMQ的C#客戶端，使用簡單方便。以下是使用C#發送消息和消費消息的代碼示例：

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queueName, false, false, false, null);
    string message = "hello rabbitmq";
    var body = Encoding.UTF8.GetBytes(message);
 
    channel.BasicPublish("", queueName, null, body);
    Console.WriteLine("Send Message：{0}", message);
}

<!-- rabbitmq 消費者 -->
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
    channel.BasicAck(ea.DeliveryTag, false);
};
channel.BasicConsume(queueName, false, consumer);

2、rabbitmq的五種消息模型

簡單隊列模型、工作隊列模型、發布/訂閱模型、路由模型、主題模型。

3、rabbitmq消息隊列5種模式

簡單隊列模型、工作隊列模型、發布/訂閱模型、路由模型、主題模型。

4、rabbitmq消息隊列原理

RabbitMQ採用AMQP協議實現，AMQP是Advanced Message Queuing Protocol的縮寫，它是一個公開的、通用的、可編程的消息傳遞協議。它的目的是為面向消息的中間件設計一個標準的應用層傳輸協議，以便於不同的消息中間件系統可以相互操作。

RabbitMQ基於AMQP1.0協議、Erlang實現，提供了高可靠性、高擴展性、高可用性、高並發性和高性能的消息中間件功能。

5、rabbitmq的四種交換機模式

Direct、Topic、Fanout、Header。

例如，使用Direct交換機模式發送消息：

<!-- rabbitmq 生產者 -->
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
    string message = "hello direct";
    var routingKey = "error";
    var body = Encoding.UTF8.GetBytes(message);
    channel.BasicPublish(exchangeName, routingKey, null, body);
 
    Console.WriteLine("Send Message：{0}", message);
}

使用Direct交換機模式消費消息：

<!-- rabbitmq 消費者 -->
var queueName = channel.QueueDeclare().QueueName;
var routingKey = "error";
channel.QueueBind(queueName, exchangeName, routingKey);
 
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine("Received message： {0}", message);
};
channel.BasicConsume(queueName, true, consumer);

四、rabbitmq使用場景

RabbitMQ可以應用於高並發、高可用、解耦系統的消息中間件。以下是RabbitMQ的常見應用場景：

1、任務異步處理：RabbitMQ能夠異步處理任務，從而提升網站性能；

2、應用解耦：RabbitMQ可以用於應用系統間的解耦；

3、數據同步：RabbitMQ能夠將數據同步到多個系統，保證數據的準確性；

4、消息通知：RabbitMQ可以作為消息通知的工具；

5、數據分析：RabbitMQ可以較好地應用於大數據分析的場景。

五、rabbitmq消息模型選取

在選擇消息模型時，需要考慮以下因素：

1、消息的類型和特性；

2、消息的生命周期和傳輸效率；

3、消息的可靠性和冪等性。

原創文章，作者：小藍，如若轉載，請註明出處：https://www.506064.com/zh-hant/n/307239.html