C#RabbitMQ消息队列的使用

C#RabbitMQ消息队列的使用

概念

1、消息(Message)

消息指的是两个应用之间传递的数据。其中数据的类型可以有很多的形式，可能只是包含文本字符串的一条消息，也可能时一个嵌入的对象。

2、消息队列(Meaasge Queue)

消息队列指的是在消息传递过程中保存消息的容器，我们可以把它理解成通讯中的缓存区，只不过是在缓存区当中可以存储多条消息数据，且提取的顺序是先进先出的原则。在消息队列中，通常还有两个必不可少的角色，也就是消息的发出是由谁来发，我们称之为生产者；以及这条消息被谁接受使用，我们称之为消费者。在消息队列当中，消费者不用考虑这条消息谁有那个生产者发送的，只负责从消息里面提取数据，甚至都可以不用管生产者是否存在，然后进行消费。消息模型图如下：

RabbitMQ的内部结构图：

在消息模型图当中，只是对消息队列的一个简单的抽象的介绍，具体到RabbitMQ当中，有几个更详细的概念也需要解释。

1、Exchange
交换器或者交换机它是负责用来接收生产者发送的消息，同时将这些消息通过指定的路由规则传输到服务器中的队列。
2、Binding
绑定，它是连接在交换器和队列的一个桥梁，一个绑定就是联接交换器和队列的路由规则，所以将交换器理解成一个由绑定构成的路由表。
3、Queue
队列，它是用来存储消息的容器，一直保留直到被消费者消费，同时呢，也是消息的终点。消息一直在队列里面，等待消费者连接到这个队列后将其消费取走。
4、Connection
网络连接，比如一个TCP连接。
5、Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
6、Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。
7、Broker
表示消息队列服务器实体。

为什么使用消息队列

从上面的描述我们可以看到，生产者和消费者之间可以说没有直接的关系，就是说在没有消费者的情况下，生产者也可以将消息传递到消息队列当中，不管生产者有没存在，消费者依旧可以从消息队列当中提取消息数据。这是一种异步的机制，那么我们在何种情况下要使用到消息队列呢？

以我们常见的订单系统为例，用户点击下了一个订单，在服务端的业务逻辑可能包含产品信息过账，库存扣减，生成相应的订单信息，还可以通知用户订单生成成功。当商家订单接到爆，处于繁忙状态，确认订单的事情被滞后了，用户不用管商家何时处理，去做自己想做的事情，等待商家自行处理订单即可。

以上是我们常见的业务解耦的情况，在需要使用到解耦的场景当中、错峰控流、广播等等，我们就可以使用消息队列。

RabbitMQ的特点

RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。
AMQP ：Advanced Message Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。
RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特点包括：

1、可靠性（Reliability）
RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。
2、灵活的路由（Flexible Routing）
在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。
3、消息集群（Clustering）
多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。
4、高可用（Highly Available Queues）
队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
5、多种协议（Multi-protocol）
RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。
6、多语言客户端（Many Clients）
RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。
7、管理界面（Management UI）
RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。
8、跟踪机制（Tracing）
如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。
9、插件机制（Plugin System）
RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

RabbitMQ安装

1、安装Erlang
由于RabbitMQ是由Erlang语言开发的，所以在安装之前，一般需要安装Erlang，大家可以去Erlang官网下载
打开链接如下图所示：大家根据自己电脑系统匹配下载框内文件。

鉴于我的电脑是64位的，下载后得到这么一个玩意，双击运行它，一直下一步即可：

安装完成后还需要配置一下环境变量：

2、安装RabbitMQ服务端

大家可以到RabbitMQ的gitHub项目中，下载window版本的服务端安装包：

下载完成后，得到这个玩意：

双击运行，一直下一步，安装完成后找到根目录：

在安装目录sbin下，运行cmd 输入 rabbitmq-plugins enable rabbitmq_management 启动界面管理服务：

启动以后，在浏览器中输入 http://127.0.0.1:15672/，进入管理页面，账户密码都是guest。
以下，介绍几个操作命令：

rabbitmqctl list_users  							查看所有用户tags
rabbitmqctl set_user_tags 用户名 administrator      给用户赋予管理员角色
rabbitmqctl change_password 用户名  密码            修改用户密码
rabbitmqctl  add_user  JC JayChou                   创建用户JC密码为JayChou
rabbitmqctl  set_permissions  JC ".*"  ".*"  ".*"   赋予JC读写所有消息队列的权限
rabbitmqctl delete_user  JC 			            删除用户JC
rabbitmq-plugins enable rabbitmq_management         开启rabbitmq_management插件，在web界面查看和管理RabbitMQ服务
rabbitmqctl status				                    查看状态

注意点：
1、rabbitmq server 默认端口是 udp的15672端口，如果是远程连接需要开启防火墙
2、guest 用户是本地账号如果，不在同一台服务器上无法连接，需要新建一个账号并且给这个账号对应的权限
以上，我们的RabbitMQ已经安装完成了，接下来，我们来介绍下.net如何使用它
3、如果需要消息持久化 ，需要把队列设置为持久化，并且每次发送消息都需要设置为持久化，重启以后会自动去加载队列以及队列的消息

RabbitMQ使用

1、导入引用
在vs中引用客户端RabbitMQ.Client，可以在官网下载，也可以在 vs的nuget中下载，需要的筒子们也可以在下面的百度云链接下载：
链接：https://pan.baidu.com/s/1KUZKkpj-f9NKL05ME6YUcQ
提取码：2wvc

2、生产者发送消息到消息队列代码：

先定义一个RabbitMQ模型，用于初始化调用

	public class RabbitMQModel
    {
        /// <summary>
        /// 远程服务器IP地址
        /// </summary>
        public string HostName { get; set; }
        /// <summary>
        /// 远程服务器端口
        /// </summary>
        public int Port { get; set; }
        /// <summary>
        /// 消息队列登录用户名
        /// </summary>
        public string UserName { get; set; }
        /// <summary>
        /// 消息队列登录密码
        /// </summary>
        public string PassWord { get; set; }
        /// <summary>
        /// 消息队列名
        /// </summary>
        public string queueName { get; set; }
        /// <summary>
        /// 交换机名
        /// </summary>
        public string exchangeName { get; set; }
    }

声明对象：using RabbitMQ.Client;

	RabbitMQModel Wcs_RabbitMQModel;
    ConnectionFactory _factoryWcsMQ;

初始化对象：

			Wcs_RabbitMQModel = new RabbitMQModel()
            {
                HostName = "localhost",
                Port = 5672,
                UserName = "lijunqing",
                PassWord = "960519",
                queueName = "WCSRabbitMQ1",
                exchangeName = "WCSRabbitMQRoute1"
            };
            if (_factoryWcsMQ == null)
            {
                _factoryWcsMQ = new ConnectionFactory
                {
                    HostName = Wcs_RabbitMQModel.HostName,
                    Port = Wcs_RabbitMQModel.Port,
                    UserName = Wcs_RabbitMQModel.UserName,
                    Password = Wcs_RabbitMQModel.PassWord
                };
            }

发送到消息队列的方法：

 #region 发送消息
        public bool SendToRPCServer(string sendContent)
        {
            if (_factoryWcsMQ != null)
            {
                IConnection conn = _factoryWcsMQ.CreateConnection();
                IModel im = conn.CreateModel();
                try
                {
                    im.ExchangeDeclare(Wcs_RabbitMQModel.exchangeName, ExchangeType.Direct);
                    im.QueueDeclare(Wcs_RabbitMQModel.queueName, true, false, false, null);
                    im.QueueBind(Wcs_RabbitMQModel.queueName, WMS_RabbitMQModel.exchangeName, ExchangeType.Direct, null);
                    Thread.Sleep(10);
                    byte[] message = Encoding.UTF8.GetBytes(sendContent);
                    im.BasicPublish(Wcs_RabbitMQModel.exchangeName, ExchangeType.Direct, null, message);
                    return true;
                }
                catch (Exception ex)
                {
                    return false;
                }
                finally
                {
                    im.Close();
                    conn.Close();
                }
            }
            return false;
        }
        #endregion

3、消费者从消息队列里面提取消息的方法：
声明对象：using RabbitMQ.Client;

		RabbitMQModel WMS_RabbitMQModel;
        ConnectionFactory _factoryWmsMQ;
        bool _Countinue = true;//用于线程循环

初始化对象：

			WMS_RabbitMQModel = new RabbitMQModel()
            {
                HostName = "localhost",
                Port = 5672,
                UserName = "lijunqing",
                PassWord = "960519",
                queueName = "WMSRabbitMQ1",
                exchangeName = "WMSRabbitMQRoute1"
            };

            if (_factoryWmsMQ == null)
            {
                _factoryWmsMQ = new ConnectionFactory
                {
                    HostName = WMS_RabbitMQModel.HostName,
                    Port = WMS_RabbitMQModel.Port,
                    UserName = WMS_RabbitMQModel.UserName,
                    Password = WMS_RabbitMQModel.PassWord
                };
            }

获取消息线程方法：

		private void WmsCommunicaFunc(object sender)
        {
            ResponseModel model = new ResponseModel();
            #region 监听接收到的数据信息
            if (WMS_RabbitMQModel != null)
            {
                ConnectionFactory factory = new ConnectionFactory
                {
                    HostName = WMS_RabbitMQModel.HostName,
                    Port = WMS_RabbitMQModel.Port,
                    UserName = WMS_RabbitMQModel.UserName,
                    Password = WMS_RabbitMQModel.PassWord
                };
                if (factory != null)
                {
                    factory.AutomaticRecoveryEnabled = true;
                    IConnection conn = factory.CreateConnection();
                    IModel im = conn.CreateModel();
                    while (_Countinue)
                    {
                        try
                        {
                            #region 数据接收
                            BasicGetResult res = im.BasicGet(WMS_RabbitMQModel.queueName, true);
                            if (res != null && res.Body.Length > 0)
                            {
                                var result = Encoding.UTF8.GetString(res.Body);
                                //消息处理逻辑代码
                            }
                            #endregion
                        }
                        catch (Exception ex)
                        {
                            if (!ex.ToString().Contains("code=404"))
                            {
                                SaveLogExceptionClient(ex.ToString());
                            }
                            else
                            {
                                #region 如果消息队列被删除/还未创建 自动创建消息队列
                                conn.Close();
                                im.Close();
                                conn = factory.CreateConnection();
                                im = conn.CreateModel();
                                im.QueueDeclare(WMS_RabbitMQModel.queueName, true, false, false, null);
                                im.ExchangeDeclare(WMS_RabbitMQModel.exchangeName, ExchangeType.Direct);
                                im.QueueBind(WMS_RabbitMQModel.queueName, WMS_RabbitMQModel.exchangeName, ExchangeType.Direct, null);
                                #endregion
                            }
                        }
                        Thread.Sleep(100);
                    }
                    im.Close();
                    conn.Close();
                }
            }
            #endregion
        }

4、可视化界面消息查看

以上，就是本次RabbitMQ的一些使用的介绍。

部分内容是从其他地方参考而来，如有侵权，烦请告知删除！谢谢