NETCore消息队列RabbitMQ实现方法与代码示例

首页

编程语言

热心网友

转载

2026-05-07

在 .NET Core 中玩转 RabbitMQ：从零搭建可靠的消息队列

消息队列是现代应用解耦和异步通信的基石，而 RabbitMQ 无疑是这个领域的明星选手。它基于 AMQP 协议，为不同应用程序间的可靠消息传递提供了强大支持。今天，我们就来深入聊聊，如何在 .NET Core 环境中，亲手搭建一个从生产到消费的完整 RabbitMQ 消息流。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

1. 安装和配置 RabbitMQ

万事开头先安家。使用 RabbitMQ 的第一步，自然是把它运行起来。最省心的方法，莫过于借助 Docker。

使用 Docker 安装 RabbitMQ

RabbitMQ 提供了官方镜像，特别是带管理界面的 management 版本，能让我们在本地快速启动一个功能齐全的服务。

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

5672 是 RabbitMQ 的核心服务端口，消息的收发都靠它。
15672 则是管理界面的 Web 端口。启动后，打开浏览器访问 https://localhost:15672，用默认账号 (guest) 和密码 (guest) 登录，就能直观地看到队列、消息等各种状态了。

服务跑起来之后，咱们的开发工作就可以正式开始了。

2. 安装 RabbitMQ 客户端库

在 .NET Core 世界里，与 RabbitMQ 打交道离不开官方的 RabbitMQ.Client 库。在项目里添加它非常简单：

dotnet add package RabbitMQ.Client

这个包囊括了连接、通道、队列声明、消息发布等所有必需的工具，是我们接下来所有操作的基础。

3. 创建生产者（Producer）

生产者，顾名思义，就是负责制造并投递消息的角色。它的核心工作流程很清晰：连接服务器、确保队列存在、然后把消息“扔”进去。

创建消息生产者代码

下面这段代码展示了一个最基本的生产者是如何工作的：

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

class Program
{
    static void Main(string[] args)
    {
        // 第一步：创建连接工厂，指向本地RabbitMQ服务器
        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：所有连接的起点，在这里配置服务器地址、凭证等信息。
QueueDeclare：这是一个幂等操作，用于确保队列存在。它是生产者和消费者能够找到彼此的“约定地点”。
BasicPublish：真正执行消息发送的方法。

参数说明：

queue: 队列的名字，这里是 "hello_queue"。
durable: 队列是否持久化。如果设为 true，即使RabbitMQ服务重启，队列也不会丢失。
exclusive: 是否为私有队列。如果设为 true，该队列仅对当前连接可见。
autoDelete: 是否自动删除。当最后一个消费者断开后，队列会被自动移除。

4. 创建消费者（Consumer）

有生产者发送消息，就得有消费者来接手处理。消费者会持续监听队列，一旦有消息到达，便触发处理逻辑。

创建消息消费者代码

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();
        }
    }
}

消费者代码的核心在于事件订阅：

EventingBasicConsumer：一个非常方便的消费者实现，通过事件通知的方式交付消息。
BasicConsume：启动消费过程。这里的 autoAck: true 意味着消息一旦被交付，就自动被视为已成功处理。

参数说明：

autoAck: 自动确认开关。设为 false 时，需要开发者手动调用 BasicAck 来确认消息处理完毕，这是实现可靠消费的关键。

5. 持久化消息

默认情况下，消息和队列都存在于内存中，服务重启就没了。对于重要消息，这可不行。如何保证消息不丢？持久化是关键。

消息持久化设置

需要在两边都做设置。首先，生产者发送消息时，要标记消息属性为持久化：

// 创建消息属性并设置为持久化
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. 消息确认机制

光把消息存下来还不够，还得确保消息被消费者成功“消化”了。这就需要将自动确认 (autoAck) 关闭，改为手动确认。

启用手动消息确认

在消费者端，我们禁用自动确认，并在业务逻辑处理成功后，显式地发出确认信号：

// 启动消费，但关闭自动确认
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 参数如果设为 true，则可以一次性确认多条消息。