【RabbitMQ】核心模型简介，以及消息的生产与消费

2025年10月7日
| 没有评论

本章学习目标

  • 理解AMQP模型中的核心概念：Connection, Channel, Producer, Consumer, Queue。

  • 创建一个.NET项目并添加RabbitMQ客户端库。

  • 使用C#编写代码发送一条消息（”Hello World”）。

  • 使用C#编写代码接收并处理这条消息。

一、理论部分

1. AMQP 0-9-1 核心模型简介

在编写代码前，我们需要理解几个核心概念，它们构成了RabbitMQ一切功能的基础：

  • 生产者 (Producer)：发送消息的应用程序。

  • 消费者 (Consumer)：接收消息的应用程序。

  • 队列 (Queue)：一个类似于邮箱的存储结构，位于RabbitMQ内部，用于存储消息。多个生产者可以向同一个队列发送消息，多个消费者也可以从同一个队列接收消息。消息只能存储在队列中。

  • 连接 (Connection)：一个TCP连接，应用程序通过它与RabbitMQ服务器建立网络连接。创建连接的开销较大。

  • 通道 (Channel)：建立在连接之上的虚拟连接。几乎所有的操作都在通道中进行。使用通道的原因是为了避免频繁创建和销毁TCP连接带来的巨大开销。一个连接可以包含多个通道。

简单工作流：Producer -> (Connection -> Channel) -> Queue -> (Channel -> Connection) -> Consumer

2. RabbitMQ.Client 库

这是RabbitMQ官方提供的.NET客户端库，它实现了AMQP协议，是我们与RabbitMQ服务器交互的桥梁。我们将通过NuGet包管理器来安装它。

二、实操部分：创建”Hello World”

我们将创建两个控制台应用程序：一个生产者（Send）和一个消费者（Receive）。

第1步：创建项目并添加NuGet包

  1. 打开IDE（如Visual Studio或VS Code），创建一个新的解决方案（Solution）。

  2. 在该解决方案中，创建两个新的控制台应用程序项目，分别命名为 Send 和 Receive。

  3. 为两个项目添加 RabbitMQ.Client NuGet包。

    • Visual Studio：右键点击项目 -> “Manage NuGet Packages…” -> 浏览 -> 搜索 RabbitMQ.Client -> 安装。

    • .NET CLI：

      cd /path/to/Send
dotnet add package RabbitMQ.Client

cd /path/to/Receive
dotnet add package RabbitMQ.Client

第2步：编写生产者（Send.cs）

将 Send 项目中的 Program.cs 替换为以下代码。请务必将 hostName、userName 和 password 替换为在上1章中设置的值（如果按教程做，应该是 localhost, myuser, mypassword）。

using System.Text;
using RabbitMQ.Client;

// 1. 创建连接工厂（ConnectionFactory）并设置连接参数
var factory = new ConnectionFactory()
{
    HostName = "localhost",     // RabbitMQ服务器地址
    UserName = "myuser",        // 用户名
    Password = "mypassword"     // 密码
};

// 2. 使用工厂创建一个连接（Connection）和一个通道（Channel）
// 'using' 语句确保在代码块结束时，连接和通道会被正确关闭和释放，这是很重要的。
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    // 3. 声明一个队列。如果队列不存在，则创建它。
    // 参数说明：
    // queue: "hello" - 队列的名称
    // durable: false - 队列是否持久化（服务器重启后是否存在）
    // exclusive: false - 是否为当前连接的专用队列（其他连接不能访问）
    // autoDelete: false - 当最后一个消费者断开后，队列是否自动删除
    channel.QueueDeclare(queue: "hello",
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    // 4. 准备要发送的消息
    string message = "Hello World!";
    var body = Encoding.UTF8.GetBytes(message); // 将消息转换为字节数组

    // 5. 发布消息到队列
    // 参数说明：
    // exchange: "" - 使用默认的（无名）交换机
    // routingKey: "hello" - 路由键，对于默认交换机，它指定了消息要发送到的队列名称
    // basicProperties: null - 消息属性（如持久化）
    // body: body - 消息体
    channel.BasicPublish(exchange: "",
                         routingKey: "hello",
                         basicProperties: null,
                         body: body);

    Console.WriteLine($" [x] Sent {message}");
}

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

代码解析：

  • 我们首先创建了一个连接工厂，并配置了连接到我们本地RabbitMQ服务器所需的参数。

  • 然后，我们建立了连接和通道。这是与RabbitMQ交互的标准方式。

  • QueueDeclare 是幂等的——它只会在队列不存在时创建它。

  • 默认交换机（""）是一个直连交换机（Direct Exchange），它会把消息路由到 routingKey 完全匹配的队列中。所以这里 routingKey: "hello" 意味着消息会被投递到名为 hello 的队列。

第3步：运行生产者并查看管理后台

  1. 运行 Send 项目（在Visual Studio中按F5，或使用CLI命令 dotnet run）。

  2. 会在控制台看到 [x] Sent Hello World!。

  3. 现在，打开RabbitMQ管理后台 (http://localhost:15672)，登录后点击 Queues 标签页。应该能看到一个名为 hello 的队列，并且它下面有 1 条消息正准备被消费（”Ready”状态）！

第4步：编写消费者（Receive.cs）

现在我们来编写消费者程序，从 hello 队列中取出消息。

将 Receive 项目中的 Program.cs 替换为以下代码：

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

// 同生产者一样，创建连接和通道
var factory = new ConnectionFactory() { HostName = "localhost", UserName = "myuser", Password = "mypassword" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    // 声明队列。这一步也是必需的，以防我们先启动消费者，而队列还不存在。
    channel.QueueDeclare(queue: "hello",
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    // 创建一个事件消费者（EventingBasicConsumer）对象，并关联到我们的通道
    var consumer = new EventingBasicConsumer(channel);

    // 当消费者收到消息时，触发这个事件
    consumer.Received += (model, ea) =>
    {
        // 消息体是字节数组，我们需要将其转换回字符串
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        Console.WriteLine($" [x] Received {message}");
    };

    // 开始消费队列中的消息
    // 参数说明：
    // queue: "hello" - 要消费的队列名称
    // autoAck: true - 自动确认模式。如果为true，消息一旦被送达，RabbitMQ会立即将其从队列中标记为删除。
    // consumer: consumer - 我们上面定义的消费者对象
    channel.BasicConsume(queue: "hello",
                         autoAck: true,
                         consumer: consumer);

    Console.WriteLine(" Press [enter] to exit.");
    Console.ReadLine(); // 保持程序运行，持续监听消息
}

代码解析：

  • 前面的连接、通道、队列声明步骤与生产者完全一致，这确保了所需的队列存在。

  • 我们创建了一个 EventingBasicConsumer 对象，并为它的 Received 事件订阅了一个处理方法。每当有消息到达时，这个匿名方法就会被调用。

  • BasicConsume 方法启动消费过程。它将我们的消费者注册到指定的队列。

  • autoAck: true 表示自动确认模式。这意味着消费者一收到消息，RabbitMQ就认为它已成功处理并立即从队列中删除该消息。这是一种简单的模式，但如果消费者在处理消息过程中崩溃，消息就会丢失。我们将在下一章学习更可靠的手动确认模式。

第5步：运行消费者

  1. 运行 Receive 项目。

  2. 会立刻在控制台看到 [x] Received Hello World!。消费者程序取走了我们之前发送的消息并打印了出来。

  3. 再次查看管理后台的 Queues 页面，会发现 hello 队列中的消息数又变回了 0。

尝试一下：

  1. 先运行 Receive 程序，让它保持运行并监听消息。

  2. 然后再运行 Send 程序多次。

  3. 观察 Receive 的控制台，它会实时地打印出每一条新收到的消息。

总结

我们已经成功实现了一个RabbitMQ应用程序。在本文中，我们：

  1. 创建了.NET项目并引入了 RabbitMQ.Client 库。

  2. 理解了AMQP模型中的 Connection、Channel、Queue、Producer 和 Consumer 等核心概念。

  3. 编写了生产者代码，成功向名为 hello 的队列发送了一条消息。

  4. 编写了消费者代码，成功从队列中取出了消息并进行处理。

  5. 使用了管理后台来验证消息的流动。

这是一个最简单的模型，生产者和消费者直接与队列打交道。在下一章，我们将学习如何实现工作队列（Work Queue）与消息确认（Ack），让多个消费者共同处理任务，这将使我们的应用变得更加实用和强大。

.net, abp