首页 > 网络编程 >.NETCore实现RabbitMQ消息队列的示例代码

.NETCore实现RabbitMQ消息队列的示例代码

来源：互联网 2026-04-08 20:08:02

在 .NET Core 中使用 RabbitMQ：构建可靠消息队列的完整指南消息队列是实现应用解耦和异步通信的关键组件，RabbitMQ 作为基于 AMQP 协议的成熟解决方案，为应用程序间的可靠消息传递提供了强大支持。本文将详细介绍如何在 .NET Core 环境中，从零开始搭建一个完整的 Ra

在 .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 管理界面。启动后访问 http://localhost:15672，使用默认账号密码（guest/guest）登录即可查看队列、消息等运行状态。

服务启动后，即可开始进行 .NET Core 客户端的开发。

2. 安装 RabbitMQ 客户端库

在 .NET Core 项目中，需要通过 NuGet 安装官方客户端库 RabbitMQ.Client。

dotnet add package RabbitMQ.Client

该库提供了连接管理、通道操作、队列声明及消息发布等核心功能，是进行 RabbitMQ 开发的基础。

3. 实现消息生产者（Producer）

生产者的主要职责是创建并向指定队列发送消息。其基本流程包括建立连接、声明队列和发布消息。

生产者代码示例

以下代码展示了一个基础生产者的实现：

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 可使队列在服务重启后依然保留。
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. 实现消息持久化

默认情况下，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 参数设为 true 时可一次性确认多条消息。