1. IT知识观首页

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]

云烟 • • 未分类 • 阅读 259

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]目前比较流行的消息队列中间件主要有:RabbitMQ， NATS， Kafka， ZeroMQ， Amazon SQS， ServiceStac

欢迎大家来到IT世界,在知识的湖畔探索吧!

注：本文首发于码友网–《.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程》

前言

在如今的互联网时代，消息队列中间件已然成为了分布式系统中重要的组件，它可以解决应用耦合，异步消息，流量削峰等应用场景。借助消息队列，我们可以实现架构的高可用、高性能、可伸缩等特性，是大型分布式系统架构中不可或缺的中间件。

目前比较流行的消息队列中间件主要有：RabbitMQ, NATS, Kafka, ZeroMQ, Amazon SQS, ServiceStack, Apache Pulsar, RocketMQ, ActiveMQ, IBM MQ等等。

本文主要为大家分享的是在.NET 5应用程序中使用消息中间件RabbitMQ的示例教程。

准备工作

在开始本文实战之前，请准备以下环境：

  • 消息中间件：RabbitMQ
  • 开发工具：Visual Studio 2019或VS Code或Rider

笔者使用的开发工具是Rider 2021.1.2。

准备解决方案和项目

创建项目

打开Rider，创建一个名为RabbitDemo的解决方案，再依次创建三个基于.NET 5的项目，分别为：RabbitDemo.Shared, RabbitDemo.Send以及RabbitDemo.Receive。

  • RabbitDemo.Shared 项目主要用于存放共用的RabbitMQ的连接相关的类；
  • RabbitDemo.Send 项目主要用于模拟生产者(发布者)；
  • RabbitDemo.Receive 项目主要用于模拟消费者(订阅者)

安装依赖包

首先，在以上创建的三个项目中分别使用包管理工具或者命令行工具安装RabbitMQ.Client依赖包，如下：

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]

编写RabbitDemo.Shared项目

RabbitDemo.Shared项目主要用于存放共用的RabbitMQ的连接相关的类。这里我们创建一个RabbitChannel类，然后在其中添加创建一些连接RabbitMQ相关的方法，包括初始化RabbitMQ的连接，关闭RabbitMQ连接等，代码如下：

using RabbitMQ.Client;

namespace RabbitDemo.Shared
{
    public class RabbitChannel
    {
        public static IModel Channel;
        private static IConnection _connection;
        public static IConnection Connection => _connection;

        public static void Init()
        {
            _connection = new ConnectionFactory
            {
                HostName = "xxxxxx", 	// 你的RabbitMQ主机地址
                UserName = "xxx",		// RabbitMQ用户名
                VirtualHost = "xxx",	// RabbitMQ虚拟主机	
                Password = "xxxxxx"		// RabbitMQ密码
            }.CreateConnection();

            Channel = _connection.CreateModel();
        }

        public static void CloseConnection()
        {
            if (Channel != null)
            {
                Channel.Close();
                Channel.Dispose();
            }

            if (_connection != null)
            {
                _connection.Close();
                _connection.Dispose();
            }
        }
    }
}

欢迎大家来到IT世界,在知识的湖畔探索吧!

编写消息生产者

在项目RabbitDemo.Send中，引用项目RabbitDemo.Shared，然后创建一个名为Send.cs类，并在其中编写生产者的代码，如下：

欢迎大家来到IT世界,在知识的湖畔探索吧!using System;
using System.Text;
using System.Threading;
using RabbitDemo.Shared;
using RabbitMQ.Client;

namespace RabbitDemo.Send
{
    public class Send
    {
        public static void Run()
        {
            for (var i = 0; i < 5; i++)
            {
                Publish(i);
                Thread.Sleep(500);
            }
        }

        private static void Publish(int index)
        {
            RabbitChannel.Channel.QueueDeclare(queue: "hello",
                durable: false,
                exclusive: false,
                autoDelete: false,
                arguments: null);

            var message = #34;Hello World from sender({index})!";
            var body = Encoding.UTF8.GetBytes(message);

            RabbitChannel.Channel.BasicPublish(exchange: "",
                routingKey: "hello",
                basicProperties: null,
                body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }
    }
}

以上示例模拟的是一个生产者一次生产了5条消息，并将消息存储到了RabbitMQ的消息队列中。

修改此项目中的Program.cs代码如下：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Send
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Send.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

编写消息消费者

在项目RabbitDemo.Receive中，引用项目RabbitDemo.Shared，然后创建一个名为Receive.cs的类，并在其中编写消费者的代码，如下：

欢迎大家来到IT世界,在知识的湖畔探索吧!using System;
using System.Linq;
using System.Text;
using RabbitDemo.Shared;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitDemo.Receive
{
    public class Receive
    {
        public static void Run()
        {
            RabbitChannel.Channel.QueueDeclare(queue: "hello",
                durable: false,
                exclusive: false,
                autoDelete: false,
                arguments: null);

            var consumer = new EventingBasicConsumer(RabbitChannel.Channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] Received {0}", message);
                RabbitChannel.Channel.BasicAck(deliveryTag:ea.DeliveryTag,multiple:false);
            };
            RabbitChannel.Channel.BasicConsume(queue: "hello",
                autoAck: false,
                consumer: consumer);
        }
    }
}

修改此项目中的Program.cs代码如下：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Receive
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Receive.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

运行

分别生成和运行生产者和消费者项目，运行效果如下：

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]

从上图可以看出，整个演示过程，RabbitMQ的消息消息是非常即时的，消费者几乎可以实时地消费生产者生产的消息。

需要确认的消息队列

在上面的生产者/消费者示例中，消息一经消费者消费，RabbitMQ会立即将消息从队列中移除。但在某些场景中，我们需要消费者确认消息被正确消费后再将其从队列中移除，RabbitMQ提供了消费确认的功能，下面我们来使用示例演示。

首先在RabbitDemo.Send中创建名为Worker.cs的类，并编写如下代码：

using System;
using System.Collections.Generic;
using System.Globalization;
using System.Text;
using System.Threading;
using RabbitDemo.Shared;
using RabbitMQ.Client;

namespace RabbitDemo.Send
{
    public class Worker
    {
        public static void Run()
        {
            for (var i = 0; i < 5; i++)
            {
                Thread.Sleep(1000);
                Publish(i);
            }
        }

        private static void Publish(int index)
        {
            var args = new Dictionary<string, object>
            {
                {"x-max-priority", 0}
            };
            RabbitChannel.Channel.QueueDeclare(queue: "task_queue",
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: args);

            var message = #34;Hello({index}) at {DateTime.Now.ToString(CultureInfo.InvariantCulture)}";
            var body = Encoding.UTF8.GetBytes(message);

            var properties = RabbitChannel.Channel.CreateBasicProperties();
            properties.Persistent = true;
            properties.Headers = new Dictionary<string, object>
            {
                {"order-no", #34;1001{index}"}
            };
            RabbitChannel.Channel.BasicPublish(exchange: "",
                routingKey: "task_queue",
                basicProperties: properties,
                body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }
    }
}

此示例模拟生和了5条消息，并将消息存放到了RabbitMQ的task_queue队列中，其中我们还通过QueueDeclare()方法的arguments参数设置了队列的优先级，也通过basicProperties参数添加了自定义的消息头(Header)参数order-no。

接着，在项目RabbitDemo.Receive项目中创建一个名为Task.cs的类，并编写如下的消费者代码：

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using RabbitDemo.Shared;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitDemo.Receive
{
    public class Task
    {
        public static void Run()
        {
            Console.WriteLine(" [*] Waiting for messages.");
            Consume();
        }

        private static void Consume()
        {
            var args = new Dictionary<string, object>
            {
                {"x-max-priority", 0}
            };
            RabbitChannel.Channel.QueueDeclare(queue: "task_queue",
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: args);
            RabbitChannel.Channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
            var consumer = new EventingBasicConsumer(RabbitChannel.Channel);
            consumer.Received += (sender, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] Received {0}", message);
                var messageBuilder = new StringBuilder();
                foreach (var headerKey in ea.BasicProperties.Headers.Keys)
                {
                    var value = ea.BasicProperties.Headers[headerKey] as byte[];
                    messageBuilder.Append("Header key: ")
                        .Append(headerKey)
                        .Append(", value: ")
                        .Append(Encoding.UTF8.GetString(value))
                        .Append("; ");
                }
                Console.WriteLine(#34;Customer properties:{messageBuilder.ToString()}");
                var sleep = 6;
                Thread.Sleep(sleep * 1000);
                Console.WriteLine(" [x] Done");
                ((EventingBasicConsumer)sender)?.Model.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            };
            RabbitChannel.Channel.BasicConsume(queue: "task_queue",
                autoAck: false,
                consumer: consumer);
        }
    }
}

在这段消费者代码中，我们主要关注的是如何向RabbitMQ确认消息的正确消费，与上面的Receive.cs消费者相比，此示例中设置了

RabbitChannel.Channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

这里的设置表示：一个消费者最多只能一次拉取1条消息

和

((EventingBasicConsumer)sender)?.Model.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

这条语句表示消息已确认被正常消费。

修改生产者的Program.cs：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Send
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Worker.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

修改消费的Program.cs：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Receive
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Task.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

效果如下：

注：本文首发于码友网–《.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程》

前言

在如今的互联网时代，消息队列中间件已然成为了分布式系统中重要的组件，它可以解决应用耦合，异步消息，流量削峰等应用场景。借助消息队列，我们可以实现架构的高可用、高性能、可伸缩等特性，是大型分布式系统架构中不可或缺的中间件。

目前比较流行的消息队列中间件主要有：RabbitMQ, NATS, Kafka, ZeroMQ, Amazon SQS, ServiceStack, Apache Pulsar, RocketMQ, ActiveMQ, IBM MQ等等。

本文主要为大家分享的是在.NET 5应用程序中使用消息中间件RabbitMQ的示例教程。

准备工作

在开始本文实战之前，请准备以下环境：

  • 消息中间件：RabbitMQ
  • 开发工具：Visual Studio 2019或VS Code或Rider

笔者使用的开发工具是Rider 2021.1.2。

准备解决方案和项目

创建项目

打开Rider，创建一个名为RabbitDemo的解决方案，再依次创建三个基于.NET 5的项目，分别为：RabbitDemo.Shared, RabbitDemo.Send以及RabbitDemo.Receive。

  • RabbitDemo.Shared 项目主要用于存放共用的RabbitMQ的连接相关的类；
  • RabbitDemo.Send 项目主要用于模拟生产者(发布者)；
  • RabbitDemo.Receive 项目主要用于模拟消费者(订阅者)

安装依赖包

首先，在以上创建的三个项目中分别使用包管理工具或者命令行工具安装RabbitMQ.Client依赖包，如下：

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]

编写RabbitDemo.Shared项目

RabbitDemo.Shared项目主要用于存放共用的RabbitMQ的连接相关的类。这里我们创建一个RabbitChannel类，然后在其中添加创建一些连接RabbitMQ相关的方法，包括初始化RabbitMQ的连接，关闭RabbitMQ连接等，代码如下：

using RabbitMQ.Client;

namespace RabbitDemo.Shared
{
    public class RabbitChannel
    {
        public static IModel Channel;
        private static IConnection _connection;
        public static IConnection Connection => _connection;

        public static void Init()
        {
            _connection = new ConnectionFactory
            {
                HostName = "xxxxxx", 	// 你的RabbitMQ主机地址
                UserName = "xxx",		// RabbitMQ用户名
                VirtualHost = "xxx",	// RabbitMQ虚拟主机	
                Password = "xxxxxx"		// RabbitMQ密码
            }.CreateConnection();

            Channel = _connection.CreateModel();
        }

        public static void CloseConnection()
        {
            if (Channel != null)
            {
                Channel.Close();
                Channel.Dispose();
            }

            if (_connection != null)
            {
                _connection.Close();
                _connection.Dispose();
            }
        }
    }
}

编写消息生产者

在项目RabbitDemo.Send中，引用项目RabbitDemo.Shared，然后创建一个名为Send.cs类，并在其中编写生产者的代码，如下：

using System;
using System.Text;
using System.Threading;
using RabbitDemo.Shared;
using RabbitMQ.Client;

namespace RabbitDemo.Send
{
    public class Send
    {
        public static void Run()
        {
            for (var i = 0; i < 5; i++)
            {
                Publish(i);
                Thread.Sleep(500);
            }
        }

        private static void Publish(int index)
        {
            RabbitChannel.Channel.QueueDeclare(queue: "hello",
                durable: false,
                exclusive: false,
                autoDelete: false,
                arguments: null);

            var message = #34;Hello World from sender({index})!";
            var body = Encoding.UTF8.GetBytes(message);

            RabbitChannel.Channel.BasicPublish(exchange: "",
                routingKey: "hello",
                basicProperties: null,
                body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }
    }
}

以上示例模拟的是一个生产者一次生产了5条消息，并将消息存储到了RabbitMQ的消息队列中。

修改此项目中的Program.cs代码如下：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Send
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Send.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

编写消息消费者

在项目RabbitDemo.Receive中，引用项目RabbitDemo.Shared，然后创建一个名为Receive.cs的类，并在其中编写消费者的代码，如下：

using System;
using System.Linq;
using System.Text;
using RabbitDemo.Shared;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitDemo.Receive
{
    public class Receive
    {
        public static void Run()
        {
            RabbitChannel.Channel.QueueDeclare(queue: "hello",
                durable: false,
                exclusive: false,
                autoDelete: false,
                arguments: null);

            var consumer = new EventingBasicConsumer(RabbitChannel.Channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] Received {0}", message);
                RabbitChannel.Channel.BasicAck(deliveryTag:ea.DeliveryTag,multiple:false);
            };
            RabbitChannel.Channel.BasicConsume(queue: "hello",
                autoAck: false,
                consumer: consumer);
        }
    }
}

修改此项目中的Program.cs代码如下：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Receive
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Receive.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

运行

分别生成和运行生产者和消费者项目，运行效果如下：

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]

从上图可以看出，整个演示过程，RabbitMQ的消息消息是非常即时的，消费者几乎可以实时地消费生产者生产的消息。

需要确认的消息队列

在上面的生产者/消费者示例中，消息一经消费者消费，RabbitMQ会立即将消息从队列中移除。但在某些场景中，我们需要消费者确认消息被正确消费后再将其从队列中移除，RabbitMQ提供了消费确认的功能，下面我们来使用示例演示。

首先在RabbitDemo.Send中创建名为Worker.cs的类，并编写如下代码：

using System;
using System.Collections.Generic;
using System.Globalization;
using System.Text;
using System.Threading;
using RabbitDemo.Shared;
using RabbitMQ.Client;

namespace RabbitDemo.Send
{
    public class Worker
    {
        public static void Run()
        {
            for (var i = 0; i < 5; i++)
            {
                Thread.Sleep(1000);
                Publish(i);
            }
        }

        private static void Publish(int index)
        {
            var args = new Dictionary<string, object>
            {
                {"x-max-priority", 0}
            };
            RabbitChannel.Channel.QueueDeclare(queue: "task_queue",
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: args);

            var message = #34;Hello({index}) at {DateTime.Now.ToString(CultureInfo.InvariantCulture)}";
            var body = Encoding.UTF8.GetBytes(message);

            var properties = RabbitChannel.Channel.CreateBasicProperties();
            properties.Persistent = true;
            properties.Headers = new Dictionary<string, object>
            {
                {"order-no", #34;1001{index}"}
            };
            RabbitChannel.Channel.BasicPublish(exchange: "",
                routingKey: "task_queue",
                basicProperties: properties,
                body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }
    }
}

此示例模拟生和了5条消息，并将消息存放到了RabbitMQ的task_queue队列中，其中我们还通过QueueDeclare()方法的arguments参数设置了队列的优先级，也通过basicProperties参数添加了自定义的消息头(Header)参数order-no。

接着，在项目RabbitDemo.Receive项目中创建一个名为Task.cs的类，并编写如下的消费者代码：

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using RabbitDemo.Shared;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitDemo.Receive
{
    public class Task
    {
        public static void Run()
        {
            Console.WriteLine(" [*] Waiting for messages.");
            Consume();
        }

        private static void Consume()
        {
            var args = new Dictionary<string, object>
            {
                {"x-max-priority", 0}
            };
            RabbitChannel.Channel.QueueDeclare(queue: "task_queue",
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: args);
            RabbitChannel.Channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
            var consumer = new EventingBasicConsumer(RabbitChannel.Channel);
            consumer.Received += (sender, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] Received {0}", message);
                var messageBuilder = new StringBuilder();
                foreach (var headerKey in ea.BasicProperties.Headers.Keys)
                {
                    var value = ea.BasicProperties.Headers[headerKey] as byte[];
                    messageBuilder.Append("Header key: ")
                        .Append(headerKey)
                        .Append(", value: ")
                        .Append(Encoding.UTF8.GetString(value))
                        .Append("; ");
                }
                Console.WriteLine(#34;Customer properties:{messageBuilder.ToString()}");
                var sleep = 6;
                Thread.Sleep(sleep * 1000);
                Console.WriteLine(" [x] Done");
                ((EventingBasicConsumer)sender)?.Model.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            };
            RabbitChannel.Channel.BasicConsume(queue: "task_queue",
                autoAck: false,
                consumer: consumer);
        }
    }
}

在这段消费者代码中，我们主要关注的是如何向RabbitMQ确认消息的正确消费，与上面的Receive.cs消费者相比，此示例中设置了

RabbitChannel.Channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

这里的设置表示：一个消费者最多只能一次拉取1条消息

和

((EventingBasicConsumer)sender)?.Model.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

这条语句表示消息已确认被正常消费。

修改生产者的Program.cs：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Send
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Worker.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

修改消费的Program.cs：

using System;
using RabbitDemo.Shared;

namespace RabbitDemo.Receive
{
    static class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("按任意键退出.");
            RabbitChannel.Init();
            Task.Run();
            Console.ReadKey();
            Console.WriteLine("正在关闭连接...");
            RabbitChannel.CloseConnection();
            Console.WriteLine("连接已关闭，退出程序.");
        }
    }
}

效果如下：

.NET 5/.NET Core应用程序中使用消息队列中间件RabbitMQ示例教程[通俗易懂]

免责声明：本站所有文章内容,图片，视频等均是来源于用户投稿和互联网及文摘转载整编而成，不代表本站观点，不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益，请在线联系站长,一经查实,本站将立刻删除。 本文来自网络,若有侵权，请联系删除，如若转载，请注明出处：https://itzsg.com/17432.html

赞 (0)
云烟云烟
0

相关推荐

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

联系我们YX

mu99908888

在线咨询： 微信交谈

邮件：itzsgw@126.com

工作时间：时刻准备着!

关注微信