当前位置：首页 > article >正文

使用 .NET Core 实现 RabbitMQ 消息队列的详细教程

article 2025/3/19 4:02:02

RabbitMQ 是一个流行的消息队列中间件，它允许应用程序通过异步消息的方式进行通信。RabbitMQ 支持 AMQP 协议，可以通过多种方式与应用程序交互。在本教程中，我们将深入探讨如何在 .NET Core 环境中使用 RabbitMQ 来实现消息队列。我们将学习如何在生产者端发送消息，消费者端接收消息，并确保消息的可靠性。

1. 安装和配置 RabbitMQ

在开始使用 RabbitMQ 之前，首先需要确保你的机器上已经安装并运行 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 后，您可以继续进行开发。

2. 安装 RabbitMQ 客户端库

在 .NET Core 中与 RabbitMQ 进行交互，我们需要使用 RabbitMQ.Client NuGet 包。可以通过以下命令在项目中添加这个依赖：

dotnet add package RabbitMQ.Client

这个库提供了与 RabbitMQ 服务进行交互所需的所有工具。

3. 创建生产者（Producer）

生产者是负责将消息发送到 RabbitMQ 的应用程序。它通过连接到 RabbitMQ 服务器、创建一个队列和交换机，将消息发布到队列中。

创建消息生产者代码

下面是一个基本的生产者示例代码，展示了如何连接到 RabbitMQ，声明队列，并发送一条简单的消息：

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

class Program
{
    static void Main(string[] args)
    {
        // 创建连接工厂
        var factory = new ConnectionFactory() { HostName = "localhost" };
        
        // 创建连接和通道
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // 声明一个队列（确保队列存在）
            channel.QueueDeclare(queue: "hello_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);

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

            // 发送消息到队列
            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 用来创建连接到 RabbitMQ 服务器的连接。
QueueDeclare 用来声明一个队列，确保队列存在。如果队列已经存在，声明将被忽略。
BasicPublish 用来将消息发送到队列。

参数说明：

queue: 队列的名称（此例中是 hello_queue）。
durable: 是否将队列标记为持久化。如果设置为 true，即使 RabbitMQ 重启，队列也会存在。
exclusive: 是否使队列只对当前连接可用。
autoDelete: 是否在最后一个消费者断开连接时自动删除队列。

4. 创建消费者（Consumer）

消费者从队列中获取并处理消息。消费者通常是另一个应用程序，它会连接到 RabbitMQ，并持续地从队列中取出消息进行处理。

创建消息消费者代码

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

class Program
{
    static void Main(string[] args)
    {
        // 创建连接工厂
        var factory = new ConnectionFactory() { HostName = "localhost" };

        // 创建连接和通道
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // 声明队列，确保消费者能够连接到相同的队列
            channel.QueueDeclare(queue: "hello_queue", 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(" [x] Received {0}", message);
            };

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

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

在上面的代码中：

QueueDeclare 用来确保消费者连接到相同的队列。
EventingBasicConsumer 是消费者的实现，用于异步接收消息。
BasicConsume 用于开始消费消息，autoAck 设置为 true，表示自动确认消息。

参数说明：

autoAck: 如果设置为 true，消费者会自动确认消息。如果设置为 false，需要手动确认消息。

5. 持久化消息

如果您希望在 RabbitMQ 重启后保持消息的持久性，可以在生产者和消费者中启用消息的持久化。

消息持久化设置

在生产者端发送持久化消息：

// 设置消息持久化
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 用于确认消息已经被处理。deliveryTag 是消息的标识符，multiple 参数表示是否确认多个消息。