RabbitMq学习笔记

Posted 2020-09-26 小羽信息

一 环境搭建

下载地址：ERLANG  http://www.erlang.org/downloads

MQ https://github.com/rabbitmq/rabbitmq-server/releases/tag/rabbitmq_v3_6_9

首先，由于RabbitMQ使用Erlang编写的，需要运行在Erlang运行时环境上，所以在安装RabbitMQ Server之前需要安装Erlang 运行时环境，可以到Erlang官网下载对应平台的安装文件。如果没有安装运行时环境，安装RabbitMQ Server的时候，会提示需要先安装Erlang环境。 安装完成之后，确保已经将Erlang的安装路径注册到系统的环境变量中。安装完Erlang之后，这个环境会自动设置，如果没有，在administrator环境下在控制台下面输入，也可以设置：

Setx  ERLANG_HOME “D:\Program Files (x86)\erl6.3″

然后，去RabbitMQ官网下载RabbitMQ Server服务端程序，选择合适的平台版本下载。安装完成之后，就可以开始使用了。

现在就可以对RabbitMQ Server进行配置了。

首先，切换到RabbitMQ Server的安装目录：

在sbin下面有很多batch文件，用来控制RabbitMQ Server，当然您也可以直接在安装开始菜单中来执行相应的操作：

最简单的方式是使RabbitMQ以Windows Service的方式在后台运行，所以我们需要以管理员权限打开cmd，然后切换到sbin目录下，执行这三条命令即可：

rabbitmq-service install

rabbitmq-service enable

rabbitmq-service start

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

下面可以使用sbin目录下面的rabbitmqctl.bat这个脚本来查看和控制服务端状态的，在cmd中直接运行rabbitmqctl status。如果看到以下结果：

显示node没有连接上，需要到C:\Windows目录下，将.erlang.cookie文件，拷贝到用户目录下 C:\Users\{用户名}，这是Erlang的Cookie文件，允许与Erlang进行交互，现在重复运行刚才的命令就会得到如下信息：

RabbitMQ Server上面也有用户概念，安装好之后，使用rabbitmqctl list_users命令，可以看到上面目前的用户：

可以看到，现在只有一个角色为administrator的名为guest的用户，这个是RabbitMQ默认为我们创建的，他有RabbitMQ的所有权限，一般的，我们需要新建一个我们自己的用户，设置密码，并授予权限，并将其设置为管理员，可以使用下面的命令来执行这一操作：

rabbitmqctl  add_user  yy  hello!

rabbitmqctl  set_permissions  yy  “.*”  “.*”  “.*”

rabbitmqctl  set_user_tags yy administrator

上面的一条命令添加了一个名为yy的用户，并设置了密码hello！，下面的命令为用户yy分别授予对所有消息队列的配置、读和写的权限。

现在我们可以将默认的guest用户删掉，使用下面的命令即可：

rabbitmqctl delete_user guest

如果要修改密码，可以使用下面的命令：

rabbitmqctl change_password {username}  {newpassowrd}

 

默认安装的Rabbit MQ 监听端口是5672

附主要命令：

名称	命令
查询服务状态	rabbitmqctl status
列举虚拟主机列表	rabbitmqctl list_vhosts
列举用户列表	rabbitmqctl list_users
添加用户和密码	rabbitmqctl  add_user  hao  abc123
设置权限	rabbitmqctl  set_permissions  yy  “.*”  “.*”  “.*”
分配用户组	rabbitmqctl  set_user_tags yy administrator
删除guest用户	rabbitmqctl delete_user guest
修改用户密码	rabbitmqctl change_password {username}  {newpassowrd}
查询交换机类型	rabbitmqctl list_exchanges
查询绑定	rabbitmqctl list_bindings
查询没有消息确认的消息	Rabbitmqctl list_queues name messages_ready messages_unacknowledged

 

 

二 WEB视图

在cmd中写出rabbitmq-plugins enable rabbitmq_management

要重启服务才能生效，可以执行

net stop RabbitMQ && net start RabbitMQ

 

完成后打开浏览器输入http://127.0.0.1:15672/#/，如果是本地的，直接用localhost（RabbitMQ 3.0之前版本端口号为55672）在输入之后，弹出登录界面，使用我们之前创建的用户登录。

账号：guest密码：guest

 

.

在该界面上可以看到当前RabbitMQServer的所有状态。

 

三 使用

在.NET中使用RabbitMQ需要下载RabbitMQ的客户端程序集，可以到官网下载，下载解压后就可以得到RabbitMQ.Client.dll，这就是RabbitMQ的客户端。

或者我们新增一个项目，在程序包管理控制器中NuGet中下载 RabbitMQ。

Install-Package RabbitMQ.Client

在代码中使用： using RabbitMQ.Client;

在使用RabitMQ之前，需要对下面的几个基本概念说明一下：

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

名称	属性	描述
生产   Producing		发送消息的程序就是一个生产者(producer)。我们一般用”P”来表示
队列   queue		邮箱的名称。消息通过你的应用程序和RabbitMQ进行传输，它们只能存储在队列（queue）中。 队列（queue）容量没有限制，你要存储多少消息都可以——基本上是一个无限的缓冲区。多个生产者（producers）能够把消息发送给同一个队列，同样，多个消费者（consumers）也能从同一个队列（queue）中获取数据。
消费   Consuming		获取消息是一样的意思。一个消费者（consumer）就是一个等待获取消息的程序
RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。     多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
vhost		一个vhost与另一个vhost是隔离开的，互不影响
measia		rabbitMQ数据存储的DB
ConnectionFactory、Connection、Channel		Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。   ConnectionFactory为Connection的制造工厂。 Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。
Message acknowledgment   消息确认		在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。   这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…
Message durability   消息持久化		如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了）
Exchange   交换机		生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）
Prefetch count   限制分发		前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。
routing key		生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。   在Exchange Type与binding key固定的情况下，我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。
Binding		RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。
Binding key		用在消费端的routingKey,为了区分BasicPublish的routingKey 在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；生产者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。 在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。   binding key 针对fanout类型的Exchange无效。将消息路由到所有绑定到该Exchange的Queue。

 

通过上面图我们可以更容易和清晰的去理解rabbitmq的工作流程。

Exchange的几种模式

名称	描述
Direct   直连	任何发送到Direct Exchange的消息都会被转发到那些binding key与routing key完全匹配的Queue中。   1.一般情况可以使用rabbitMQ自带的Exchange：”(该Exchange的名字为空字符串，下文称其为default Exchange)。 2.这种模式下不需要将Exchange进行任何绑定(binding)操作 3.消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。 4.如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。 以上图的配置为例， 1)routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）； 2)routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。
Headers   管道	headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。   在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。 该类型的Exchange没有用到过
Topic   局部	任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上   1.这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。 2.这种模式需要RouteKey，也许要提前绑定Exchange与Queue。 3.在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。 4.“#”表示0个或若干个关键字，“*”表示一个关键字。如“log.*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。 5.同样，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息 以上图中的配置为例， 1)routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2， 2) routingKey=”lazy.orange.fox”的消息会路由到Q1，Q2 3) routingKey=”lazy.brown.fox”的消息会路由到Q2， 4) routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）； 5)routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。
Fanout   分发 ,多播	任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。   1.可以理解为路由表的模式 2.这种模式不需要RouteKey 3.这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。 4.如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。 上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

 

 

Producter:

//声明交换机： channel.ExchangeDeclare(ExchangeName, “direct”, durable: true, autoDelete: false, arguments: null);//声明队列： channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null);//绑定队列   channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);    // 内容的基本属性   var props = channel.CreateBasicProperties();//设置内容的持久化     props.Persistent = true; //或者properties.DeliveryMode = 2; //消息持久化 //公平分发、同一时间只处理一个消息。 channel.BasicQos(0, 1, false); //发布消息channel.BasicPublish(exchange: ExchangeName, routingKey: QueueName, basicProperties: props, body: msgBody);

consumer：

//声明队列   channel.QueueDeclare(QueueName, durable: true, autoDelete: false, exclusive: false, arguments: null); //channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//告诉broker同一时间只处理一个消息//channel.QueueBind(QueueName, ExchangeName, routingKey: QueueName);//定义消费者  var consumer = new EventingBasicConsumer(channel);//接收消息  consumer.Received += (model, ea) =>      {                                   var msgBody = Encoding.UTF8.GetString(ea.Body);            Console.WriteLine(string.Format(“***接收时间:{0}，消息内容：{1}”, DateTime.Now.ToString(“yyyy-MM-dd HH:mm:ss”), msgBody));       };//消费消息 channel.BasicConsume(QueueName, noAck: true, consumer: consumer);

RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。

但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。
RabbitMQ基础概念详细介绍

RabbitMQ中实现RPC的机制是：

• 客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
• 服务器端收到消息并处理
• 服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
• 客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理