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

五月花号 發表於 2025-9-20 23:59:00

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

<h3>本章学习目标</h3>
<ul>
<li>
理解AMQP模型中的核心概念：Connection, Channel, Producer, Consumer, Queue。
</li>
</ul>
<ul>
<li>
创建一个.NET项目并添加RabbitMQ客户端库。
</li>
<li>
使用C#编写代码发送一条消息（"Hello World"）。
</li>
<li>
使用C#编写代码接收并处理这条消息。
</li>
</ul>
<hr>
<h2>一、理论部分</h2>
<h3>1. AMQP 0-9-1 核心模型简介</h3>
在编写代码前，我们需要理解几个核心概念，它们构成了RabbitMQ一切功能的基础：
<ul>
<li>
生产者 (Producer)：发送消息的应用程序。
</li>
<li>
消费者 (Consumer)：接收消息的应用程序。
</li>
<li>
队列 (Queue)：一个类似于邮箱的存储结构，位于RabbitMQ内部，用于存储消息。多个生产者可以向同一个队列发送消息，多个消费者也可以从同一个队列接收消息。消息只能存储在队列中。
</li>
<li>
连接 (Connection)：一个TCP连接，应用程序通过它与RabbitMQ服务器建立网络连接。创建连接的开销较大。
</li>
<li>
通道 (Channel)：建立在连接之上的虚拟连接。几乎所有的操作都在通道中进行。使用通道的原因是为了避免频繁创建和销毁TCP连接带来的巨大开销。一个连接可以包含多个通道。
</li>
</ul>
简单工作流：<code>Producer</code> -> (<code>Connection</code> -> <code>Channel</code>) -> <code>Queue</code> -> (<code>Channel</code> -> <code>Connection</code>) -> <code>Consumer</code>
<h3>2. <code>RabbitMQ.Client</code> 库</h3>
这是RabbitMQ官方提供的.NET客户端库，它实现了AMQP协议，是我们与RabbitMQ服务器交互的桥梁。我们将通过NuGet包管理器来安装它。
<hr>
<h2>二、实操部分：创建"Hello World"</h2>
我们将创建两个控制台应用程序：一个生产者（Send）和一个消费者（Receive）。
<h3>第1步：创建项目并添加NuGet包</h3>
<ol start="1">
<li>
打开IDE（如Visual Studio或VS Code），创建一个新的解决方案（Solution）。
</li>
<li>
在该解决方案中，创建两个新的控制台应用程序项目，分别命名为 <code>Send</code> 和 <code>Receive</code>。
</li>
<li>
为两个项目添加 <code>RabbitMQ.Client</code> NuGet包。
<ul>
<li>
Visual Studio：右键点击项目 -> "Manage NuGet Packages..." -> 浏览 -> 搜索 <code>RabbitMQ.Client</code> -> 安装。
</li>
<li>
.NET CLI：
<div class="md-code-block md-code-block-light">
<div class="md-code-block-banner-wrap">
<div class="md-code-block-banner md-code-block-banner-lite">
<div class="_121d384">
<div class="d2a24f03">
<div class="cnblogs_code">
<pre>cd /path/to/Send
dotnet add package RabbitMQ.Client

cd /path/to/Receive
dotnet add package RabbitMQ.Client</pre>
</div>
</div>
</div>
</div>
</div>
</div>
</li>
</ul>
</li>
</ol>
<h3>第2步：编写生产者（Send.cs）</h3>
将 <code>Send</code> 项目中的 <code>Program.cs</code> 替换为以下代码。请务必将 <code>hostName</code>、<code>userName</code> 和 <code>password</code> 替换为在上1章中设置的值（如果按教程做，应该是 <code>localhost</code>, <code>myuser</code>, <code>mypassword</code>）。
<div class="md-code-block md-code-block-light">
<div class="md-code-block-banner-wrap">
<div class="md-code-block-banner md-code-block-banner-lite">
<div class="_121d384">
<div class="d2a24f03">
<div class="cnblogs_code">
<pre>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($" Sent {message}");
}

Console.WriteLine(" Press to exit.");
Console.ReadLine();</pre>
</div>
</div>
</div>
</div>
</div>
</div>
代码解析：
<ul>
<li>
我们首先创建了一个连接工厂，并配置了连接到我们本地RabbitMQ服务器所需的参数。
</li>
<li>
然后，我们建立了连接和通道。这是与RabbitMQ交互的标准方式。
</li>
<li>
<code>QueueDeclare</code> 是幂等的——它只会在队列不存在时创建它。
</li>
<li>
默认交换机（<code>""</code>）是一个直连交换机（Direct Exchange），它会把消息路由到 <code>routingKey</code> 完全匹配的队列中。所以这里 <code>routingKey: "hello"</code> 意味着消息会被投递到名为 <code>hello</code> 的队列。
</li>
</ul>
<h3>第3步：运行生产者并查看管理后台</h3>
<ol start="1">
<li>
运行 <code>Send</code> 项目（在Visual Studio中按F5，或使用CLI命令 <code>dotnet run</code>）。
</li>
<li>
会在控制台看到 <code> Sent Hello World!</code>。
</li>
<li>
现在，打开RabbitMQ管理后台 (http://localhost:15672)，登录后点击 Queues 标签页。应该能看到一个名为 <code>hello</code> 的队列，并且它下面有 1 条消息正准备被消费（"Ready"状态）！ 

</li>

</ol>
<h3>第4步：编写消费者（Receive.cs）</h3>
现在我们来编写消费者程序，从 <code>hello</code> 队列中取出消息。
将 <code>Receive</code> 项目中的 <code>Program.cs</code> 替换为以下代码：
<div class="cnblogs_code">
<pre>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($" Received {message}");
};

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

Console.WriteLine(" Press to exit.");
Console.ReadLine(); // 保持程序运行，持续监听消息
}</pre>
</div>
代码解析：
<ul>
<li>
前面的连接、通道、队列声明步骤与生产者完全一致，这确保了所需的队列存在。
</li>
<li>
我们创建了一个 <code>EventingBasicConsumer</code> 对象，并为它的 <code>Received</code> 事件订阅了一个处理方法。每当有消息到达时，这个匿名方法就会被调用。
</li>
<li>
<code>BasicConsume</code> 方法启动消费过程。它将我们的消费者注册到指定的队列。
</li>
<li>
<code>autoAck: true</code> 表示自动确认模式。这意味着消费者一收到消息，RabbitMQ就认为它已成功处理并立即从队列中删除该消息。这是一种简单的模式，但如果消费者在处理消息过程中崩溃，消息就会丢失。我们将在下一章学习更可靠的手动确认模式。
</li>
</ul>
<h3>第5步：运行消费者</h3>
<ol start="1">
<li>
运行 <code>Receive</code> 项目。
</li>
<li>
会立刻在控制台看到 <code> Received Hello World!</code>。消费者程序取走了我们之前发送的消息并打印了出来。
</li>
<li>
再次查看管理后台的 Queues 页面，会发现 <code>hello</code> 队列中的消息数又变回了 0。
</li>
</ol>
尝试一下：
<ol start="1">
<li>
先运行 <code>Receive</code> 程序，让它保持运行并监听消息。
</li>
<li>
然后再运行 <code>Send</code> 程序多次。
</li>
<li>
观察 <code>Receive</code> 的控制台，它会实时地打印出每一条新收到的消息。
</li>
</ol><hr>
<h2>总结</h2>
我们已经成功实现了一个RabbitMQ应用程序。在本文中，我们：
<ol start="1">
<li>
创建了.NET项目并引入了 <code>RabbitMQ.Client</code> 库。
</li>
<li>
理解了AMQP模型中的 <code>Connection</code>、<code>Channel</code>、<code>Queue</code>、<code>Producer</code> 和 <code>Consumer</code> 等核心概念。
</li>
<li>
编写了生产者代码，成功向名为 <code>hello</code> 的队列发送了一条消息。
</li>
<li>
编写了消费者代码，成功从队列中取出了消息并进行处理。
</li>
<li>
使用了管理后台来验证消息的流动。
</li>
</ol>
这是一个最简单的模型，生产者和消费者直接与队列打交道。在下一章，我们将学习如何实现工作队列（Work Queue）与消息确认（Ack），让多个消费者共同处理任务，这将使我们的应用变得更加实用和强大。 
来源：https://www.cnblogs.com/jixingsuiyuan/p/19103083

頁: [1]

琼殿技术论坛's Archiver

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