.NET Core實現(xiàn)RabbitMQ消息隊列的示例代碼

更新時間：2025年05月31日 09:30:42 作者：江沉晚呤時

RabbitMQ支持AMQP協(xié)議,可以通過多種方式與應用程序交互,本文主要介紹了.NET Core實現(xiàn)RabbitMQ消息隊列的示例代碼,實現(xiàn)在生產(chǎn)者端發(fā)送消息,消費者端接收消息,并確保消息的可靠性

RabbitMQ 是一個流行的消息隊列中間件，它允許應用程序通過異步消息的方式進行通信。RabbitMQ 支持 AMQP 協(xié)議，可以通過多種方式與應用程序交互。在本教程中，我們將深入探討如何在 .NET Core 環(huán)境中使用 RabbitMQ 來實現(xiàn)消息隊列。我們將學習如何在生產(chǎn)者端發(fā)送消息，消費者端接收消息，并確保消息的可靠性。

1. 安裝和配置 RabbitMQ

在開始使用 RabbitMQ 之前，首先需要確保你的機器上已經(jīng)安裝并運行 RabbitMQ?？梢酝ㄟ^以下方式安裝 RabbitMQ：

使用 Docker 安裝 RabbitMQ

RabbitMQ 提供了官方的 Docker 鏡像，這使得在本地機器上運行 RabbitMQ 非常簡單。

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

5672 是 RabbitMQ 的默認消息隊列端口。
15672 是 RabbitMQ 管理插件的 Web 界面端口。通過瀏覽器訪問 http://localhost:15672 可以登錄 RabbitMQ 管理界面，默認的用戶名和密碼都是 guest。

安裝并啟動 RabbitMQ 后，您可以繼續(xù)進行開發(fā)。

2. 安裝 RabbitMQ 客戶端庫

在 .NET Core 中與 RabbitMQ 進行交互，我們需要使用 RabbitMQ.Client NuGet 包?？梢酝ㄟ^以下命令在項目中添加這個依賴：

dotnet add package RabbitMQ.Client

這個庫提供了與 RabbitMQ 服務進行交互所需的所有工具。

3. 創(chuàng)建生產(chǎn)者（Producer）

生產(chǎn)者是負責將消息發(fā)送到 RabbitMQ 的應用程序。它通過連接到 RabbitMQ 服務器、創(chuàng)建一個隊列和交換機，將消息發(fā)布到隊列中。

創(chuàng)建消息生產(chǎn)者代碼

下面是一個基本的生產(chǎn)者示例代碼，展示了如何連接到 RabbitMQ，聲明隊列，并發(fā)送一條簡單的消息：

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

class Program
{
    static void Main(string[] args)
    {
        // 創(chuàng)建連接工廠
        var factory = new ConnectionFactory() { HostName = "localhost" };
        
        // 創(chuàng)建連接和通道
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // 聲明一個隊列（確保隊列存在）
            channel.QueueDeclare(queue: "hello_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);

            // 創(chuàng)建消息
            string message = "Hello, RabbitMQ!";
            var body = Encoding.UTF8.GetBytes(message);

            // 發(fā)送消息到隊列
            channel.BasicPublish(exchange: "", routingKey: "hello_queue", basicProperties: null, body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

在上面的代碼中：

ConnectionFactory 用來創(chuàng)建連接到 RabbitMQ 服務器的連接。
QueueDeclare 用來聲明一個隊列，確保隊列存在。如果隊列已經(jīng)存在，聲明將被忽略。
BasicPublish 用來將消息發(fā)送到隊列。

參數(shù)說明：

queue: 隊列的名稱（此例中是 hello_queue）。
durable: 是否將隊列標記為持久化。如果設置為 true，即使 RabbitMQ 重啟，隊列也會存在。
exclusive: 是否使隊列只對當前連接可用。
autoDelete: 是否在最后一個消費者斷開連接時自動刪除隊列。

4. 創(chuàng)建消費者（Consumer）

消費者從隊列中獲取并處理消息。消費者通常是另一個應用程序，它會連接到 RabbitMQ，并持續(xù)地從隊列中取出消息進行處理。

創(chuàng)建消息消費者代碼

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

class Program
{
    static void Main(string[] args)
    {
        // 創(chuàng)建連接工廠
        var factory = new ConnectionFactory() { HostName = "localhost" };

        // 創(chuàng)建連接和通道
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // 聲明隊列，確保消費者能夠連接到相同的隊列
            channel.QueueDeclare(queue: "hello_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);

            // 創(chuàng)建消費者對象
            var consumer = new EventingBasicConsumer(channel);

            // 消息處理邏輯
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] Received {0}", message);
            };

            // 開始消費消息
            channel.BasicConsume(queue: "hello_queue", autoAck: true, consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }
}

在上面的代碼中：

QueueDeclare 用來確保消費者連接到相同的隊列。
EventingBasicConsumer 是消費者的實現(xiàn)，用于異步接收消息。
BasicConsume 用于開始消費消息，autoAck 設置為 true，表示自動確認消息。

參數(shù)說明：

autoAck: 如果設置為 true，消費者會自動確認消息。如果設置為 false，需要手動確認消息。

5. 持久化消息

如果您希望在 RabbitMQ 重啟后保持消息的持久性，可以在生產(chǎn)者和消費者中啟用消息的持久化。

消息持久化設置

在生產(chǎn)者端發(fā)送持久化消息：

// 設置消息持久化
var properties = channel.CreateBasicProperties();
properties.Persistent = true; // 設置消息為持久化

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

此外，聲明隊列時也需要設置 durable: true，確保隊列本身是持久化的。

channel.QueueDeclare(queue: "hello_queue", durable: true, exclusive: false, autoDelete: false, arguments: null);

6. 消息確認機制

在消息傳遞過程中，為了確保消息被成功消費并避免丟失，可以啟用消息確認機制。在這種情況下，消費者需要顯式確認消息。

啟用手動消息確認

在消費者端禁用自動確認，并手動確認每條已成功處理的消息：

channel.BasicConsume(queue: "hello_queue", autoAck: false, consumer: consumer);

consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine(" [x] Received {0}", message);
    
    // 手動確認消息
    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};

BasicAck 用于確認消息已經(jīng)被處理。deliveryTag 是消息的標識符，multiple 參數(shù)表示是否確認多個消息。