C#调用RabbitMQ实现消息队列

文章目录

前言
消息队列
准备工作
安装rabbitmq-server
RabbitMQ应用
- KibaRabbitMQSend
- KibaRabbitMQReceived
服务器端应用
- 安装Erlang环境
- 安装RabbitMQ
添加用户和权限
- RabbitMQ的简单指令
参考文献

前言

前几天在做日志收集，用到了RabbitMQ，它作为一种中间件，需要对其进行下载，安装，和配置。

消息队列

什么是消息队列？，我们这样想一下，用户访问网站，最终是要将数据以HTTP的协议的方式，通过网络传输到主机的某个端口上的。

那么，接收数据的方式是什么呢？自然是端口监听啦。

那消息队列是什么就很好解释了

它就是端口监听，接到数据后，将数据排列起来。

RabbitMQ是一个非常流行的，开源的，使用Erlang语言编写的框架，通常在电信级平台中作为消息中间件使用，RabbitMQ实现了高级的AMQP协议用于实现进程间，应用程序间，服务器之间的消息交互，而且它还有一个非常好的特性，你可以使用自定义插件来扩展RabbitMQ的功能，而且它还支持多协议，高性能，高可靠，集群以及高可用队列。

当你在用 RabbitMQ 时，你要理解下面两个术语

队列是一个数据结构上的概念，支持 FIFO 特性，在本文中，消息队列就是一个可以存放消息的巨大缓存。
producer(生产者) 的使命是生成数据并推送到队列的一种角色组件，consumer(消费者)它能够从存储消息的队列中提取数据进行消费，生产者-消费者是并行编程中非常流行的设计模式之一

创建队列的方式也多种多样，你可以编码创建，也可以通过管理员用户界面，甚至通过 PowerShell 进行队列创建。

那这件事，我们不用中间件能做吗？

当然能做啦，写个TCP/UDP/Socket的软件就可以做啦。

如下图所示：

既然自己可以做消息队列，那为什么要用RabbitMQ?

因为，RabbitMQ成熟的开源中间件，可靠性有保证，bug少，性能也非常好。

而C#代码默认是使用托管内存的，所以，想写出媲美RabbitMQ性能的消息队列，就必须离开我们常用的托管内存，使用非托管内存，但这个代价就太大了；而且最终能否达到RabbitMQ的性能水平还是个未知数。
还有就是RabbitMQ除了基础的消息队列管理，还有很多很强大的额外功能，而自己开发消息队列，很难如此尽善尽美。

我们还会发现，在消息队列里有很多概念，什么消息总线啊，什么工作队列啊等等。

要怎么理解这些概念呢？

很简单，不要去理解。这些概念其实是人家代码架构的模式，不要去理解他们，【记】就完了，人家的中间件就是按照这个模式工作的。

比如，我写了一个接收消息的总控制器，然后我为他命名为总线，那这个控制器就是总线，没有理由，这就是定义。

准备工作

首先，我们访问官网https://www.rabbitmq.com，点击Get Started。

然后，网站会自动跳转到当前首页Get Started的锚点位置，如下图：

然后我们点击DownLoad+Installation，进入到下载界面。

在下载页面中，我们找到安装指南，然后再点击官网推荐的Windows系统的安装包，如下图：

现在，我们进入了Windows安装指南界面了。

首先，我们看一下预览信息，如下图：

在预览里，我们得知，安装RabbitMQ有两种方法，一种是使用Chocolatey安装，一种是使用官方安装包安装。

Chocolatey是什么呢?随手百度一下，原来他是一个软件包管理工具，也就是说，Chocolatey是类似于Nuget的一种工具。

由于Chocolatey的使用，我不是很熟悉，所以，这里选择使用官方安装包安装。
点击【Using the official installer】，我们进入了【Using the official installer】对应的锚点，如下图。

找到安装包，然后下载，这样就完成了rabbitmq的安装。

除了下载安装包，我们还会发现，在【Using the official installer】段落里，有提醒我们，RabbitMQ是有依赖的，依赖一个Erlang语言的框架（类似于C#语言的NetFramework）。

点击【Erlang/OTP Version Tree】进去下载，这里我们选择对应win64位进行下载和安装。

首先，我们运行otp_win64_24.3.4.exe，安装依赖框架Erlang\OTP。

安装完成后，设置环境变量如下：

然后运行CMD，输入erl，测试安装是否成功，如下图：

安装成功。

安装rabbitmq-server

安装完依赖后，我们接着安装rabbitmq-server-3.10.1.exe。

从这个文件名上，我们发现了一个问题，那就是，我们即将安装的RabbitMQ，是一个服务端啊。

什么？服务端？难道还有客户端？？？

其实这也很好理解，想一下最开始我举的那个例子，消息队列是需要一个监听端口的服务端的，然后客户端向这个服务端发送请求。

这样是不是就很好的理解RabbitMQ了呢

安装完RabbitMQ服务端后，我们还是启动CMD，用命令行来查看下安装状态。

首先输入下面的命令，将路径定位到RabbitMQ的sbin路径下,然后输入rabbitmqctl status查看状态。

启动管理工具的命令行

rabbitmq-plugins enable rabbitmq_management

启动成功后，在浏览器输入地址http://127.0.0.1:15672/，进入管理页面，账户密码都是guest

RabbitMQ还有很多常用命令，大家可以自行百度。

到此，RabbitMQ服务端的环境配置好了，正常情况，这些配置应该在服务器进行，但我为了测试方便，就把服务端也安装在本机了，因此我下面调用RabbitMQ时，连接的主机IP都是localhost。

RabbitMQ应用

首先在VS中创建两个控制台应用程序，KibaRabbitMQSend和KibaRabbitMQReceived。

然后引入RabbitMQ的开源类库。

在C#里使用RabbitMQ开源类库非常简单，可以去官网下载一个.NET版本的RabbitMQ客户端类库，也可以直接在VS的工具—>Nuget管理包上搜索RabbitMQ，然后安装，如下图：

KibaRabbitMQSend

安装完RabbitMQ开源类库后，我们编写代码，实现向RabbitMQ服务器发送消息，代码如下：

channel 常用于和 server 进行通讯从而发送和接收消息，不过上面这种队列是不持久的，为啥这么说呢？因为我在 QueueDeclare 方法的第二个参数中设置了 false，所以送往这个 queue 的 message 只会存留于内存中，一旦 server 重启，这个数据将会丢失。

static void Main(string[] args)
{var factory = new ConnectionFactory();factory.HostName = "localhost";//主机名，Rabbit会拿这个IP生成一个endpoint，这个很熟悉吧，就是socket绑定的那个终结点。factory.UserName = "guest";//默认用户名,用户可以在服务端自定义创建，有相关命令行factory.Password = "guest";//默认密码using (var connection = factory.CreateConnection())//连接服务器，即正在创建终结点。{//创建一个通道，这个就是Rabbit自己定义的规则了，如果自己写消息队列，这个就可以开脑洞设计了//这里Rabbit的玩法就是一个通道channel下包含多个队列Queueusing (var channel = connection.CreateModel()){channel.QueueDeclare("kibaQueue", false, false, false, null);//创建一个名称为kibaqueue的消息队列var properties = channel.CreateBasicProperties();properties.DeliveryMode = 1;string message = "I am Kiba518"; //传递的消息内容channel.BasicPublish("", "kibaQueue", properties, Encoding.UTF8.GetBytes(message)); //生产消息Console.WriteLine($"Send:{message}");}}
}

在VS运行代码后，然后我们使用命令行rabbitmqctl list_queues，去RabbitMQ的服务器查看当前消息队列，如下图：

可以看到，我们的消息已经发送成功了。

KibaRabbitMQReceived

现在我们编写接收消息代码，如下：

static void Main(string[] args)
{var factory = new ConnectionFactory();factory.HostName = "localhost";factory.UserName = "guest";factory.Password = "guest";using (var connection = factory.CreateConnection()){using (var channel = connection.CreateModel()){channel.QueueDeclare("kibaQueue", false, false, false, null);/* 这里定义了一个消费者，用于消费服务器接受的消息* C#开发需要注意下这里，在一些非面向对象和面向对象比较差的语言中，是非常重视这种设计模式的。* 比如RabbitMQ使用了生产者与消费者模式，然后很多相关的使用文章都在拿这个生产者和消费者来表述。* 但是，在C#里，生产者与消费者对我们而言，根本算不上一种设计模式，他就是一种最基础的代码编写规则。* 所以，大家不要复杂的名词吓到，其实，并没那么复杂。* 这里，其实就是定义一个EventingBasicConsumer类型的对象，然后该对象有个Received事件，* 该事件会在服务接收到数据时触发。*/var consumer = new EventingBasicConsumer(channel);//消费者channel.BasicConsume("kibaQueue", true, consumer);//消费消息consumer.Received += (model, ea) =>{var body = ea.Body;var message = Encoding.UTF8.GetString(body);};}}
}

值得一提的是：RabbitMQ 对持久化也提供了非常好的支持，有两种模式的 queue 可供选择：持久化和非持久化，如果采用持久化模式，消息是存放在硬盘中的，反之非持久化的模式，数据仅仅存放于内存中，一旦server重启，非持久化模式的队列数据将会丢失，最后补充一下，持久化可以用于以下三个级别上： Queue，Exchange 和 Message。

运行代码,然后我们使用命令行rabbitmqctl list_queues，去RabbitMQ的服务器查看当前消息队列，如下图：

可以看到，消息已经被使用了。
现在我们在发送代码出做一个for循环，看看消息接收速度是什么样的，代码如下，for循环了100次，每次间隔3秒。

for (int i = 0; i < 100; i++)
{channel.QueueDeclare("kibaQueue", false, false, false, null);//创建一个名称为kibaQueue的消息队列var properties = channel.CreateBasicProperties();properties.DeliveryMode = 1;string message = "I am Kiba518"; //传递的消息内容channel.BasicPublish("", "kibaQueue", properties, Encoding.UTF8.GetBytes(message)); //生产消息Console.WriteLine($"Send:{message}");Thread.Sleep(3000);
}

效果如下：

可以看到，发送消息和接收消息，几乎是同步的，效果非常理想。

服务器端应用

在上文，我们的RabbitMQ服务是安装在我的本机上的；现在我们把服务移植到服务器上，然后再来测试一下。

在服务器端安装RabbitMQ和在本机安装的步骤是一样的，但是安装完成后，我们需要设置下防火墙的入站规则和出站规则，将5672的UDP端口开放一下。

为什么要开放端口是5672？因为RabbitMQ的默认的消息接收和发送端口就是5672。

安装Erlang环境

安装GCC GCC-C++ Openssl等模块,安装过就不需要安装了

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

安装ncurses

yum -y install ncurses-devel

安装erlang环境

wget http://erlang.org/download/otp_src_18.2.1.tar.gz
tar xvfz otp_src_18.2.1.tar.gz
./configure
make install

安装RabbitMQ

下载rabbitmq-server-3.6.9.tar.xz

wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.9/rabbitmq-server-generic-unix-3.6.9.tar.xz

如果上述地址失效，可以用备份链接下载：
https://gakkiwife.lanzoub.com/i2JU205bg1uj，解压后，在文件尾部加上tar.xz

对于下载xz包进行解压，首先先下载xz压缩工具：

yum install xz

对rabbitmq包进行解压：

xz -d xz -d rabbitmq-server-generic-unix-3.6.9.tar.xz
tar -xvf rabbitmq-server-generic-unix-3.6.9.tar

随后移动至/usr/local/下改名rabbitmq：

cp -r rabbitmq_server-3.6.9 /usr/local/rabbitmq

这种下载的方式解压后直接可以使用，无需再编译安装；
进入到rabbit文件内，其命令文件存在于sbin文件夹下，因此需要将sbin文件夹的路径添加到PATH中：修改/etc/profile

export PATH=/usr/local/rabbitmq/sbin:$PATH

执行source /etc/profile使得PATH路径更新，rabbitMQ安装成功。

随后启用MQ管理方式：

rabbitmq-plugins enable rabbitmq_management   #启动后台管理
rabbitmq-server -detached    #后台运行rabbitmq

设置端口号，可供外部访问：

iptables -I INPUT -p tcp --dport 15672 -j ACCEPT

添加用户和权限

默认网页guest用户是不允许访问的，需要增加一个用户修改一下权限，代码如下：添加用户:`rabbitmqctl add_user admin admin`添加权限:`rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"`修改用户角色:`rabbitmqctl set_user_tags admin administrator`

然后就可以远程访问了，然后可直接配置用户权限等信息。

RabbitMQ的简单指令

启动服务：rabbitmq-server -detached【 /usr/local/rabbitmq/sbin/rabbitmq-server  -detached 】
重启服务：rabbitmq-server restart
关闭服务：rabbitmqctl stop
查看状态：rabbitmqctl status
列出角色：rabbitmqctl list_users
开启某个插件：rabbitmq-pluginsenable xxx
关闭某个插件：rabbitmq-pluginsdisablexxx
注意：重启服务器后生效。

参考文献

https://blog.51cto.com/lhxsoft/3182393
https://github.com/kiba518/KibaRabbitMQ
https://segmentfault.com/a/1190000021539070
https://segmentfault.com/a/1190000038580754
RabbitMQgege版本下载