.NET 使用RabbitMQ

叁歲伎倆 2022-05-21 13:44 172阅读 0赞

# 1、说明 #

在企业应用系统领域，会面对不同系统之间的通信、集成与整合，尤其当面临异构系统时，这种分布式的调用与通信变得越发重要。其次，系统中一般会有很多对实时性要求不高的但是执行起来比较较耗时的地方，比如发送短信，邮件提醒，更新文章阅读计数，记录用户操作日志等等，如果实时处理的话，在用户访问量比较大的情况下，对系统压力比较大。

面对这些问题，我们一般会将这些请求，放在消息队列MQ中处理；异构系统之间使用消息进行通讯。

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取或者订阅队列中的消息。

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。

消息传递相较文件传递与远程过程调用（RPC）而言，似乎更胜一筹，因为它具有更好的平台无关性，并能够很好地支持并发与异步调用。所以如果系统中出现了如下情况:

对操作的实时性要求不高，而需要执行的任务极为耗时；  
存在异构系统间的整合；  
　　一般的可以考虑引入消息队列。对于第一种情况，常常会选择消息队列来处理执行时间较长的任务。引入的消息队列就成了消息处理的缓冲区。消息队列引入的异步通信机制，使得发送方和接收方都不用等待对方返回成功消息，就可以继续执行下面的代码，从而提高了数据处理的能力。尤其是当访问量和数据流量较大的情况下，就可以结合消息队列与后台任务，通过避开高峰期对大数据进行处理，就可以有效降低数据库处理数据的负荷。

# 2、搭建环境 #

2.1 安装Erlang语言运行环境

由于RabbitMQ使用Erlang语言编写，所以先安装Erlang语言运行环境。具体请看博客：[windows配置Erlang环境][windows_Erlang]

2.2 安装RabbitMQ服务端

地址 [http://www.rabbitmq.com/][http_www.rabbitmq.com]

使RabbitMQ以Windows Service的方式在后台运行：打开cmd切换到sbin目录下执行

我自己的安装位置  
![这里写图片描述][70]

现在RabbitMQ的服务端已经启动起来了。

要查看和控制RabbitMQ服务端的状态，可以用rabbitmqctl这个脚本。

比如查看状态：

rabbitmqctl status

![这里写图片描述][70 1]

遇到下面问题：

![这里写图片描述][70 2]

解决办法

a. 需要到C:\\Windows目录下，将.erlang.cookie文件，拷贝到用户目录下 C:\\Users\{用户名\}，这是Erlang的Cookie文件，允许与Erlang进行交互。

b.登录账号权限问题

![这里写图片描述][70 3]

*记得重启一下*

使用命令查看用户：rabbitmqctl list\_users

![这里写图片描述][70 4]

RabbitMQ会为我们创建默认的用户名guest和密码guest，guest默认拥有RabbitMQ的所有权限。

一般的，我们需要新建一个我们自己的用户，设置密码，并授予权限，并将其设置为管理员，可以使用下面的命令来执行这一操作：

rabbitmqctl  add_user  admin 123456   //创建用户admin密码为123456
    rabbitmqctl  set_permissions  admin ".*"  ".*"  ".*"    //赋予admin读写所有消息队列的权限
    rabbitmqctl  set_user_tags admin administrator    //分配用户组

修改admin密码为123：

rabbitmqctl change_password admin123

删除用户admin：

rabbitmqctl delete_user  admin

也可以开启rabbitmq\_management插件，在web界面查看和管理RabbitMQ服务

rabbitmq-plugins enable rabbitmq_management

**下载RabbitMQ的Client端dll**

下载地址：[http://www.rabbitmq.com/releases/rabbitmq-dotnet-client/][http_www.rabbitmq.com_releases_rabbitmq-dotnet-client]  
![这里写图片描述][70 5]  
![这里写图片描述][70 6]  
解压，我们需要的是这个文件，以后会引用到vs的项目中：  
![这里写图片描述][70 7]

# 3、使用 #

## 3.1在使用RabitMQ之前，先对几个概念做一下说明 ##

RabbitMQ是一个消息代理。他从消息生产者(producers)那里接收消息，然后把消息送给消息消费者（consumer）在发送和接受之间，他能够根据设置的规则进行路由，缓存和持久化。

一般提到RabbitMQ和消息，都用到一些专有名词。

生产(Producing)意思就是发送。发送消息的程序就是一个生产者(producer)。我们一般用”P”来表示：

队列(queue)就是邮箱的名称。消息通过你的应用程序和RabbitMQ进行传输，它们只能存储在队列（queue）中。 队列（queue）容量没有限制，你要存储多少消息都可以——基本上是一个无限的缓冲区。多个生产者（producers）能够把消息发送给同一个队列，同样，多个消费者（consumers）也能从同一个队列（queue）中获取数据。

消费（Consuming）和获取消息是一样的意思。一个消费者（consumer）就是一个等待获取消息的程序。我们把它画作”C”：

通常，消息生产者，消息消费者和消息代理不在同一台机器上。

## 3.2 Hello Word ##

![这里写图片描述][70 8]  
**3.2.1 首先创建名为Send的控制台项目,需要引用RabbitMQ.Client.dll。这个程序作为Producer生产者，用来发送数据：**

using RabbitMQ.Client;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    
    namespace Send { 
        class Program { 
            static void Main(string[] args)
            {
                var factory = new ConnectionFactory();
                factory.HostName = "localhost";
                factory.UserName = "admin";
                factory.Password = "123456";
    
                using(var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        //创建一个名称为hello的消息队列
                        channel.QueueDeclare("hello", false, false, false, null);
                        //内容
                        string message = "RibbitMQ";
                        var body = Encoding.UTF8.GetBytes(message);
                        //开始传递
                        channel.BasicPublish("", "hello", null, body);
                        Console.WriteLine("已发送：{0}", message);
                        Console.ReadKey();
                    }
                }
            }
        }
    }

首先，需要创建一个ConnectionFactory，设置目标，由于是在本机，所以设置为localhost，如果RabbitMQ不在本机，只需要设置目标机器的IP地址或者机器名称即可，然后设置前面创建的用户名和密码。

紧接着要创建一个Channel，如果要发送消息，需要创建一个队列，然后将消息发布到这个队列中。在创建队列的时候，只有RabbitMQ上该队列不存在，才会去创建。消息是以二进制数组的形式传输的，所以如果消息是实体对象的话，需要序列化和然后转化为二进制数组。

现在客户端发送代码已经写好了，运行之后，消息会发布到RabbitMQ的消息队列中，现在需要编写服务端的代码连接到RabbitMQ上去获取这些消息。

**3.2.2创建名为Receive的控制台项目，引用RabbitMQ.Client.dll。作为Consumer消费者，用来接收数据：**

using RabbitMQ.Client;
    using RabbitMQ.Client.Events;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    
    namespace Receive { 
        class Program { 
            static void Main(string[] args)
            {
                var factory = new ConnectionFactory();
                factory.HostName = "localhost";
                factory.UserName = "admin";
                factory.Password = "123456";
    
                using (var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        //创建一个名称为hello的消息队列
                        channel.QueueDeclare("hello", false, false, false, null);
    
                        var consumer = new EventingBasicConsumer(channel);
    
                        channel.BasicConsume("hello", false, consumer);
    
                        consumer.Received += (model, e) =>
                        {
                            var body = e.Body;
                            //内容
                            var message = Encoding.UTF8.GetString(body);
    
                            Console.WriteLine("已接收：{0}", message);
                        };
                        Console.ReadKey();
                    }
                }
            }
        }
    }

和发送一样，首先需要定义连接，然后声明消息队列。要接收消息，需要定义一个Consume，然后在接收消息的事件中处理数据。

**3.2.3 运行结果**  
![这里写图片描述][70 9]

![这里写图片描述][70 10]

## 3.3 工作队列 ##

前面的例子展示了如何在指定的消息队列发送和接收消息。

现在我们创建一个工作队列（work queue）来将一些耗时的任务分发给多个工作者（workers）：  
![这里写图片描述][70 11]  
　工作队列（work queues, 又称任务队列Task Queues）的主要思想是为了避免立即执行并等待一些占用大量资源、时间的操作完成。而是把任务（Task）当作消息发送到队列中，稍后处理。一个运行在后台的工作者（worker）进程就会取出任务然后处理。当运行多个工作者（workers）时，任务会在它们之间共享。

这个在网络应用中非常有用，它可以在短暂的HTTP请求中处理一些复杂的任务。在一些实时性要求不太高的地方，我们可以处理完主要操作之后，以消息的方式来处理其他的不紧要的操作，比如写日志等等。

现在发送一些字符串，把这些字符串当作复杂的任务。这里使用time.sleep()函数来模拟耗时的任务。在字符串中加上点号（.）来表示任务的复杂程度，一个点（.）将会耗时1秒钟。比如”Hello…”就会耗时3秒钟。

**send端**

using RabbitMQ.Client;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    
    namespace Send1 { 
        class Program { 
            static void Main(string[] args)
            {
                var factory = new ConnectionFactory();
                factory.HostName = "localhost";
                factory.UserName = "admin";
                factory.Password = "123456";
    
                using (var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        channel.QueueDeclare("hello", false, false, false, null);
                        string message = GetMessage(args);
                        var properties = channel.CreateBasicProperties();
                        properties.DeliveryMode = 2;
    
                        var body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish("", "hello", properties, body);
                        Console.WriteLine("已发送：{0}", message);
    
                    }
                }
                Console.ReadKey();
            }
            private static string GetMessage(string[] args)
            {
                return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
            }
        }
    }

**receive端**

using RabbitMQ.Client;
    using RabbitMQ.Client.Events;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    
    namespace Receive1 { 
        class Program { 
            static void Main(string[] args)
            {
                var factory = new ConnectionFactory();
                factory.HostName = "localhost";
                factory.UserName = "admin";
                factory.Password = "123456";
    
                using (var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        channel.QueueDeclare("hello", false, false, false, null);
                        var consumer = new QueueingBasicConsumer(channel);
                        channel.BasicConsume("hello", true, consumer);
    
                        while (true)
                        {
                            var e = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                            var body = e.Body;
                            //内容
                            var message = Encoding.UTF8.GetString(body);
    
                            var dots = message.Split('.').Length - 1
    ;
                            Thread.Sleep(dots * 1000);
                            Console.WriteLine("已接收：{0}", message);
                        }
    
                    }
                }
            }
        }
    }

**轮询分发**

使用工作队列的一个好处就是它能够并行的处理队列。如果堆积了很多任务，我们只需要添加更多的工作者（workers）就可以了，扩展很简单。

现在，我们先启动两个接收端，等待接受消息，然后启动一个发送端开始发送消息。  
![这里写图片描述][70 12]  
![这里写图片描述][70 13]

这里的结果不是想要的结果

## 3.4 消息响应 ##

当处理一个比较耗时得任务的时候，也许想知道消费者（consumers）是否运行到一半就挂掉。在当前的代码中，当RabbitMQ将消息发送给消费者（consumers）之后，马上就会将该消息从队列中移除。此时，如果把处理这个消息的工作者（worker）停掉，正在处理的这条消息就会丢失。同时，所有发送到这个工作者的还没有处理的消息都会丢失。

我们不想丢失任何任务消息。如果一个工作者（worker）挂掉了，我们希望该消息会重新发送给其他的工作者（worker）。

为了防止消息丢失，RabbitMQ提供了消息响应（acknowledgments）机制。消费者会通过一个ack（响应），告诉RabbitMQ已经收到并处理了某条消息，然后RabbitMQ才会释放并删除这条消息。

如果消费者（consumer）挂掉了，没有发送响应，RabbitMQ就会认为消息没有被完全处理，然后重新发送给其他消费者（consumer）。这样，即使工作者（workers）偶尔的挂掉，也不会丢失消息。

消息是没有超时这个概念的；当工作者与它断开连的时候，RabbitMQ会重新发送消息。这样在处理一个耗时非常长的消息任务的时候就不会出问题了。

消息响应默认是开启的。在之前的例子中使用了*no\_ack=True*标识把它关闭。是时候移除这个标识了，当工作者（worker）完成了任务，就发送一个响应。

channel.BasicConsume("hello", false, consumer);
    
    while (true)
    {
        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
    
        var body = ea.Body;
        var message = Encoding.UTF8.GetString(body);
    
        int dots = message.Split('.').Length - 1;
        Thread.Sleep(dots * 1000);
    
        Console.WriteLine("Received {0}", message);
        Console.WriteLine("Done");
    
        channel.BasicAck(ea.DeliveryTag, false);
    }

## 3.5 消息持久化 ##

前面已经搞定了即使消费者down掉，任务也不会丢失，但是，如果RabbitMQ Server停掉了，那么这些消息还是会丢失。

当RabbitMQ Server 关闭或者崩溃，那么里面存储的队列和消息默认是不会保存下来的。如果要让RabbitMQ保存住消息，需要在两个地方同时设置：需要保证队列和消息都是持久化的。

首先，要保证RabbitMQ不会丢失队列，所以要做如下设置：

bool durable = true;
    channel.QueueDeclare("hello", durable, false, false, null);

虽然在语法上是正确的，但是在目前阶段是不正确的，因为我们之前已经定义了一个非持久化的hello队列。RabbitMQ不允许我们使用不同的参数重新定义一个已经存在的同名队列，如果这样做就会报错。现在，定义另外一个不同名称的队列：

bool durable = true;
    channel.queueDeclare("task_queue", durable, false, false, null);

queueDeclare 这个改动需要在发送端和接收端同时设置。

现在保证了task\_queue这个消息队列即使在RabbitMQ Server重启之后，队列也不会丢失。 然后需要保证消息也是持久化的， 这可以通过设置IBasicProperties.SetPersistent 为true来实现：

var properties = channel.CreateBasicProperties();
    properties.SetPersistent(true);

需要注意的是，将消息设置为持久化并不能完全保证消息不丢失。虽然他告诉RabbitMQ将消息保存到磁盘上，但是在RabbitMQ接收到消息和将其保存到磁盘上这之间仍然有一个小的时间窗口。 RabbitMQ 可能只是将消息保存到了缓存中，并没有将其写入到磁盘上。持久化是不能够一定保证的，但是对于一个简单任务队列来说已经足够。如果需要消息队列持久化的强保证，可以使用publisher confirms

## 3.6 公平分发 ##

你可能会注意到，消息的分发可能并没有如我们想要的那样公平分配。比如，对于两个工作者。当奇数个消息的任务比较重，但是偶数个消息任务比较轻时，奇数个工作者始终处理忙碌状态，而偶数个工作者始终处理空闲状态。但是RabbitMQ并不知道这些，他仍然会平均依次的分发消息。

为了改变这一状态，我们可以使用basicQos方法，设置perfetchCount=1 。这样就告诉RabbitMQ 不要在同一时间给一个工作者发送多于1个的消息，或者换句话说。在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。

channel.BasicQos(0, 1, false);

原文博客：[https://www.cnblogs.com/longlongogo/p/6489574.html][https_www.cnblogs.com_longlongogo_p_6489574.html]

[windows_Erlang]: https://blog.csdn.net/hyunbar/article/details/80959659
[http_www.rabbitmq.com]: http://www.rabbitmq.com/
[70]: /images/20220521/31fc1a1d47c6435db69e27c0bd48d714.png
[70 1]: /images/20220521/dab33efddcd949fdafff49622c1d382d.png
[70 2]: /images/20220521/718ba0a0ddd84627bfb58a937711c29c.png
[70 3]: /images/20220521/8a920873088b4032888e9af0a0c6ebcc.png
[70 4]: /images/20220521/ff2a29756301444180201d63740e8381.png
[http_www.rabbitmq.com_releases_rabbitmq-dotnet-client]: http://www.rabbitmq.com/releases/rabbitmq-dotnet-client/
[70 5]: /images/20220521/a58027f1820c46b9bb746c0a5aaa0e03.png
[70 6]: /images/20220521/24862c4bcb774e85afc9567d69a5cc96.png
[70 7]: /images/20220521/9b3eabcc0e13473db52e51c46f44f47f.png
[70 8]: /images/20220521/1b99bad47194459a8a591f22c0aba339.png
[70 9]: /images/20220521/d8eab3e672ae41a083509cc74914faa0.png
[70 10]: /images/20220521/ce17a3faf5e04d2e9bc5b9b575fee334.png
[70 11]: /images/20220521/e74753e068304a8ba0447a023a022936.png
[70 12]: /images/20220521/02afb5a59b5449ea9a4b04bf33def3fc.png
[70 13]: /images/20220521/0f757b211cf24c9eaeab431363ffd27c.png
[https_www.cnblogs.com_longlongogo_p_6489574.html]: https://www.cnblogs.com/longlongogo/p/6489574.html