有缘友爱 發表於 2025-5-31 09:30:42

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

<div id="navCategory"><h5 class="catalogue">目录</h5><ul class="first_class_ul"><li><a href="#_label0">1. 安装和配置 RabbitMQ</a></li><ul class="second_class_ul"><li><a href="#_lab2_0_0">使用 Docker 安装 RabbitMQ</a></li></ul><li><a href="#_label1">2. 安装 RabbitMQ 客户端库</a></li><ul class="second_class_ul"></ul><li><a href="#_label2">3. 创建生产者（Producer）</a></li><ul class="second_class_ul"><li><a href="#_lab2_2_1">创建消息生产者代码</a></li><li><a href="#_lab2_2_2">参数说明：</a></li></ul><li><a href="#_label3">4. 创建消费者（Consumer）</a></li><ul class="second_class_ul"><li><a href="#_lab2_3_3">创建消息消费者代码</a></li><li><a href="#_lab2_3_4">参数说明：</a></li></ul><li><a href="#_label4">5. 持久化消息</a></li><ul class="second_class_ul"><li><a href="#_lab2_4_5">消息持久化设置</a></li></ul><li><a href="#_label5">6. 消息确认机制</a></li><ul class="second_class_ul"><li><a href="#_lab2_5_6">启用手动消息确认</a></li></ul><li><a href="#_label6">7. 运行和测试</a></li><ul class="second_class_ul"></ul><li><a href="#_label7">8. 总结</a></li><ul class="second_class_ul"></ul></ul></div><p>RabbitMQ 是一个流行的消息队列中间件，它允许应用程序通过异步消息的方式进行通信。RabbitMQ 支持 AMQP 协议，可以通过多种方式与应用程序交互。在本教程中，我们将深入探讨如何在 .NET Core 环境中使用 RabbitMQ 来实现消息队列。我们将学习如何在生产者端发送消息，消费者端接收消息，并确保消息的可靠性。</p>
<p class="maodian"><a name="_label0"></a></p><h2>1. 安装和配置 RabbitMQ</h2>
<p>在开始使用 RabbitMQ 之前，首先需要确保你的机器上已经安装并运行 RabbitMQ。可以通过以下方式安装 RabbitMQ：</p>
<p class="maodian"><a name="_lab2_0_0"></a></p><h3>使用 Docker 安装 RabbitMQ</h3>
<p>RabbitMQ 提供了官方的 Docker 镜像，这使得在本地机器上运行 RabbitMQ 非常简单。</p>
<div class="jb51code"><pre class="brush:bash;">docker pull rabbitmq:management
docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:management
</pre></div>
<ul><li><code>5672</code>&nbsp;是 RabbitMQ 的默认消息队列端口。</li><li><code>15672</code>&nbsp;是 RabbitMQ 管理插件的 Web 界面端口。通过浏览器访问&nbsp;<code>http://localhost:15672</code>&nbsp;可以登录 RabbitMQ 管理界面，默认的用户名和密码都是&nbsp;<code>guest</code>。</li></ul>
<p>安装并启动 RabbitMQ 后，您可以继续进行开发。</p>
<p class="maodian"><a name="_label1"></a></p><h2>2. 安装 RabbitMQ 客户端库</h2>
<p>在 .NET Core 中与 RabbitMQ 进行交互，我们需要使用&nbsp;<code>RabbitMQ.Client</code>&nbsp;NuGet 包。可以通过以下命令在项目中添加这个依赖：</p>
<div class="jb51code"><pre class="brush:csharp;">dotnet add package RabbitMQ.Client
</pre></div>
<p>这个库提供了与 RabbitMQ 服务进行交互所需的所有工具。</p>
<p class="maodian"><a name="_label2"></a></p><h2>3. 创建生产者（Producer）</h2>
<p>生产者是负责将消息发送到 RabbitMQ 的应用程序。它通过连接到 RabbitMQ 服务器、创建一个队列和交换机，将消息发布到队列中。</p>
<p class="maodian"><a name="_lab2_2_1"></a></p><h3>创建消息生产者代码</h3>
<p>下面是一个基本的生产者示例代码，展示了如何连接到 RabbitMQ，声明队列，并发送一条简单的消息：</p>
<div class="jb51code"><pre class="brush:csharp;">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(" Sent {0}", message);
      }

      Console.WriteLine(" Press to exit.");
      Console.ReadLine();
    }
}
</pre></div>
<p>在上面的代码中：</p>
<ul><li><code>ConnectionFactory</code>&nbsp;用来创建连接到 RabbitMQ 服务器的连接。</li><li><code>QueueDeclare</code>&nbsp;用来声明一个队列，确保队列存在。如果队列已经存在，声明将被忽略。</li><li><code>BasicPublish</code>&nbsp;用来将消息发送到队列。</li></ul>
<p class="maodian"><a name="_lab2_2_2"></a></p><p class="maodian"><a name="_lab2_3_4"></a></p><h3>参数说明：</h3>
<ul><li><code>queue</code>: 队列的名称（此例中是&nbsp;<code>hello_queue</code>）。</li><li><code>durable</code>: 是否将队列标记为持久化。如果设置为&nbsp;<code>true</code>，即使 RabbitMQ 重启，队列也会存在。</li><li><code>exclusive</code>: 是否使队列只对当前连接可用。</li><li><code>autoDelete</code>: 是否在最后一个消费者断开连接时自动删除队列。</li></ul>
<p class="maodian"><a name="_label3"></a></p><h2>4. 创建消费者（Consumer）</h2>
<p>消费者从队列中获取并处理消息。消费者通常是另一个应用程序，它会连接到 RabbitMQ，并持续地从队列中取出消息进行处理。</p>
<p class="maodian"><a name="_lab2_3_3"></a></p><h3>创建消息消费者代码</h3>
<div class="jb51code"><pre class="brush:csharp;">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) =&gt;
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" Received {0}", message);
            };

            // 开始消费消息
            channel.BasicConsume(queue: "hello_queue", autoAck: true, consumer: consumer);

            Console.WriteLine(" Press to exit.");
            Console.ReadLine();
      }
    }
}
</pre></div>
<p>在上面的代码中：</p>
<ul><li><code>QueueDeclare</code>&nbsp;用来确保消费者连接到相同的队列。</li><li><code>EventingBasicConsumer</code>&nbsp;是消费者的实现，用于异步接收消息。</li><li><code>BasicConsume</code>&nbsp;用于开始消费消息，<code>autoAck</code>&nbsp;设置为&nbsp;<code>true</code>，表示自动确认消息。</li></ul>
<h3>参数说明：</h3>
<ul><li><code>autoAck</code>: 如果设置为&nbsp;<code>true</code>，消费者会自动确认消息。如果设置为&nbsp;<code>false</code>，需要手动确认消息。</li></ul>
<p class="maodian"><a name="_label4"></a></p><h2>5. 持久化消息</h2>
<p>如果您希望在 RabbitMQ 重启后保持消息的持久性，可以在生产者和消费者中启用消息的持久化。</p>
<p class="maodian"><a name="_lab2_4_5"></a></p><h3>消息持久化设置</h3>
<p>在生产者端发送持久化消息：</p>
<div class="jb51code"><pre class="brush:csharp;">// 设置消息持久化
var properties = channel.CreateBasicProperties();
properties.Persistent = true; // 设置消息为持久化

channel.BasicPublish(exchange: "", routingKey: "hello_queue", basicProperties: properties, body: body);
</pre></div>
<p>此外，声明队列时也需要设置&nbsp;<code>durable: true</code>，确保队列本身是持久化的。</p>
<div class="jb51code"><pre class="brush:csharp;">channel.QueueDeclare(queue: "hello_queue", durable: true, exclusive: false, autoDelete: false, arguments: null);
</pre></div>
<p class="maodian"><a name="_label5"></a></p><h2>6. 消息确认机制</h2>
<p>在消息传递过程中，为了确保消息被成功消费并避免丢失，可以启用消息确认机制。在这种情况下，消费者需要显式确认消息。</p>
<p class="maodian"><a name="_lab2_5_6"></a></p><h3>启用手动消息确认</h3>
<p>在消费者端禁用自动确认，并手动确认每条已成功处理的消息：</p>
<div class="jb51code"><pre class="brush:csharp;">channel.BasicConsume(queue: "hello_queue", autoAck: false, consumer: consumer);

consumer.Received += (model, ea) =&gt;
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine(" Received {0}", message);
   
    // 手动确认消息
    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
</pre></div>
<p><code>BasicAck</code>&nbsp;用于确认消息已经被处理。<code>deliveryTag</code>&nbsp;是消息的标识符，<code>multiple</code>&nbsp;参数表示是否确认多个消息。</p>
<p class="maodian"><a name="_label6"></a></p><h2>7. 运行和测试</h2>
<ul><li>启动消费者应用程序，确保它可以连接到 RabbitMQ 并等待消息。</li><li>启动生产者应用程序，它将发送消息到 RabbitMQ 队列。</li><li>消费者将从队列中接收到消息，并进行处理。</li></ul>
<p>如果一切配置正确，您将在控制台中看到生产者发送的消息以及消费者处理的消息。</p>
<p class="maodian"><a name="_label7"></a></p><h2>8. 总结</h2>
<p>通过本教程，我们学习了如何在 .NET Core 中使用 RabbitMQ 实现一个简单的消息队列系统。关键步骤包括：</p>
<ul><li>安装 RabbitMQ 客户端库。</li><li>在生产者中声明队列并发送消息。</li><li>在消费者中声明队列并处理消息。</li><li>配置消息持久化和确认机制，确保消息的可靠性。</li></ul>
<p>RabbitMQ 是一个强大的消息队列中间件，适用于各种需要解耦和异步通信的应用程序。通过灵活的交换机和队列配置，您可以实现不同的消息传递模式，以满足不同的业务需求。</p>
頁: [1]
查看完整版本: .NET Core实现RabbitMQ消息队列的示例代码