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

在 .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) 登录，就能直观地看到队列、消息等各种状态了。

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

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，则可以一次性确认多条消息。

7. 运行和测试

• 首先，启动消费者应用程序。它会保持运行，静静等待消息的到来。
• 然后，运行生产者应用程序。你会看到控制台输出发送成功的提示。
• 瞬间，消费者的控制台就会打印出接收到的消息内容。

如果一切顺利，你就完成了一次完整的异步消息传递。这个过程清晰地展示了生产与消费的解耦。

8. 总结

走完这一趟，我们在 .NET Core 中实现 RabbitMQ 消息队列的核心路径就清晰了：

• 环境搭建：用 Docker 快速部署 RabbitMQ 服务和管理界面。
• 基础通信：通过 RabbitMQ.Client 库创建生产者和消费者，完成最简单的消息收发。
• 可靠性加固：通过队列与消息的持久化设置，防止服务器重启导致数据丢失；再结合手动消息确认机制，确保消息被业务逻辑成功处理。

RabbitMQ 的强大远不止于此，其灵活的路由规则、多样的交换机类型，足以支撑起从简单任务队列到复杂事件驱动架构的各种场景。理解并掌握这些基础步骤，无疑是构建更健壮、更松耦合的分布式系统的良好开端。