RabbitMQ入门

Posted 2021-01-09 zj-dandelion

RabbitMQ

一、 Windows下安装RabbitMQ及常用命令

RabbitMQ是一个在AMQP协议标准基础上完整的，可服用的企业消息系统。它遵循Mozilla Public License开源协议，采用Erlang实现的工业级消息队列服务器，RabbitMQ是建立在Erlang OTP平台上。

1. 安装Erlang
   在安装RabbitMQ之前，需要先安装Erlang ，安装地址http://www.erlang.org/download.html。
   安装成功后配置环境变量ERLANG_HOME=D:Program Fileserl10.1
2. 安装RabbitMQ
   官网下载地址http://www.rabbitmq.com/download.html
   需要注意的是：默认安装RabbitMQ监听端口是5672

3. 配置

   1. 激活RabbitMQ’s Management Plugin

      使用RabbitMQ管理插件，可以更好的可视化方式查看RabbitMQ服务器实例的状态。
      CMD输入命令"D:Program FilesRabbitMQ Server abbitmq_server-3.7.8sbin abbitmq-plugins.bat" enable rabbitmq_management
      
      插件安装成功之后，需要重启服务net stop RabbitMQ && net start RabbitMQ
      
      可能会发生错误：”发生系统错误5”即不是系统管理员权限
      问题解决方案：使用管理员打开cmd再次执行命令
      

   2. 创建用户，密码，绑定角色

      使用rabbitmqctl控制台命令，来创建用户，密码，绑定权限等
      RabbitMQ的用户管理包括增加用户，删除用户，查看用户列表，修改用户密码。

• 查看已有用户及用户角色rabbitmqctl.bat list_users
  
• 新增一个用户rabbitmqctl.bat add_user username password
  
  查询用户及角色 会发现 [administrator] ，此为RabbitMQ的角色，RabbitMQ可分为5类：超级管理员，监控者，策略制定者，普通管理员以及其他。
  • 超级管理员（administrator）
    • 可登陆管理控制台（启用management plugin的情况下），可查看所有的信息，并且可以对用户，策略（policy）进行操作
  • 监控者（monitoring）
    • 可登陆管理控制台（启用management plugin的情况下），同时可以查看rabbitmq节点的相关信息（进程数，内存使用情况，磁盘使用情况等）
  • 策略制定者（policymaker）
    • 可登陆管理控制台（启用management plugin的情况下），同时可以对policy进行管理。
  • 普通管理者（management）
    • 仅可登陆管理控制台（启用management plugin的情况下），无法看到节点信息，也无法对策略进行管理。
  • 其他
    • 无法登陆管理控制台，通常就是普通的消费者和生产者。
• 给用户添加角色rabbitmqctl.bat set_user_tags [user] [role]
  
  除了给用户添加administrator角色之外，还可以添加monitoring，policymaker，management，自定义名称等角色。
  此外还可以给一个用户设置多个角色rabbitmqctl.bat set_user_tags [user] [role1] [role2]...
  
• 修改用户密码rabbitmqctl.bat change_password [user] [password]
• 删除用户rabbitmqctl.bat delete_user [user]

  1. VirtualHost与权限管理

• VirtualHost
  • 像mysql有数据库的概念并且可以指定用户对库和表等操作的权限。 RabbitMQ也是有类似的权限管理的。在RabbitMQ中可以虚拟消息服务器VirtualHost，每个VirtualHost都是一个相对独立的RabbitMQ服务器，每个VirtualHost之间都是相互隔离的。exchange、queue、message不能互通。
  • 在RabbitMQ中无法通过AMQP创建VirtualHost，可以通过命令来创建rabbitmqctl add_vhost [vhostname]，也可以通过管理控制台来创建。如下图
• 权限管理
  • 主要分为三部分：创建用户、分配角色、设置权限（本节主要讲解设置权限）
  • 配置权限 set_permissions [-p <vhostpath>] <user> <conf> <write> <read>

通过http://localhost:15672登陆管理控制台，可进行查看信息、监控，策略等操作。

二、RabbitMQ的初步了解

1. 消息怎么投递的（生产者消息投递都是面向交换机的）
   RabbitMQ是面向交换机投递消息的。交换机可能绑定有许多队列，那么交换机如何将消息投递给这些队列？
   面向交换机设计的优势：
   • 借助了数据链路层那个交换机的设计思想。除了层级分明外，还能从分提高链路利用率。
   • 从代码层面来看：如果没有交换机，你至少得维护一个十分庞大的路由表，然后从路由表正确投递消息。有了交换机，这里路由表就会被拆分到多个交换机里面。
   • 高度解耦，不同的交换机可有不同的路由规则
     在RabbitMQ，交换机有4中投递方式，就是枚举类BuiltinExchangeType的4个枚举变量：
   • DIRECT:会将所有消息先取消息的ROUTE_KEY,然后投递到与ROUTE_KEY绑定的队列里面(if(msg.routekey.equals(queue.routekey)))
   • FANOUT:此种模式下，根本不检查消息的ROUTE_KEY，直接投送到交换机所拥有的所有队列里
   • TOPIC:
   • HEADERS:
2. 消费者怎么消费消息（消费者消费消息是面向消息队列的，这与生成者有点不一样）
   channel.basicConsume(QUEUE_AUTODELETE, true, consumer);consumer是Consumer类的一个实例。
3. 消费传递是否可靠
   可靠。
4. 消息的声明周期
   一旦受到消费者应答，标识消息已被消费，则消息被回收掉
5. 队列声明周期
   channel.queueDeclare(QUEUE_NAME, false, false, true, null);
   第二个参数设置为null，会将消息持久化到磁盘第四个参数设置为true表示没有消息并且没有连接则删除该队列

MQ的相关名称

1. ConnectionFactory、connection、Channel
   此三者都是RabbitMQ对外提供的API中最基本的对象
   Connection是RabbitMQ的Socket链接，它封装了Socket协议相关部分逻辑。
   ConnectionFactory为Connection的制造工厂。
   Channel是我们与RabbitMQ打交道的最重要的接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。
2. Queue
   Queue（队列）是RabbitMQ的内部对象，用于存储消息。
   RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者可以从Queue中获取消息并消费。
   多个消费者可以订阅一个Queue，这时Queue中的消息会被平均分摊给多个消费者并进行处理，而不是每个消费者都收到所有的消息并处理。
3. Message acknowledgment
   在实际应用中，可能会发生消费者收到Queue中的消息，但是没有处理完成就宕机（或出现其他意外）情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将消息从RabbitMQ中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。
   这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug—Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑。
   另外，pub message是没有ack的。
4. Message Durability
   如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分的情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。
5. Prefetch count
   前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。
6. Exchange
   我们了解到生产者将消息投递到Queue中，实际上这在RabbitMQ中永远都不会发生。实际情况是，生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

• Exchange是按照什么逻辑将消息路由到Queue的？会在8、Binding中介绍
• RabbitMQ中的Exchange由四种类型，不同的类型有着不同的路由策略，这将在10、Exchange Types中介绍

1. Routing key
   生产者将消息发送给Exchange的时候，一般会指定一个Routing key，来指定这个消息的路由规则，而这个Routing key需要与Exchange Type及Binding key联合使用才能最终生效。
   在Exchange Type 和 Binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。RabbitMQ为routing key设定的长度限制为255bytes。
2. Binding
   RabbitMQ中通过Binding将Exchange和Queue关联起来，这样RabbitMQ就知道如何正确的将消息路由到指定的Queue了。
3. Binding key

• 在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；
• 消费者将消息发送给Exchange时，一般会指定一个routing key；
• 当binding key与routing key相匹配时，消息将会被路由到对应的 Queue中，
  在绑定多个Queue到同一个Exchange的时候，这些binding允许使用相同的binding key。binding key并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视Binding key，而是将消息路由到所有绑定到该Exchange的Queue。

1. Exchange Types
   RabbitMQ常用的Exchange Type有fanout，direct，topic，headers这四种（AMQP规范里还提到了两种Exchange Type，分别为system和自定义，这里不予描述）

• fanout
  • fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与他绑定的Queue中。
• direct
  • direct会把消息路由到那些binding key与routing key完全匹配的Queue中。
• topic
  • 前面讲到direct类型的Exchange路由规则是完全匹配binding key和routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：routing key为一个句点号.分隔的字符串（我们将被句点号“.”分隔开的每一段独立的字符串称为一个单词）。比如stock.usd.nyse、nyse.vmw、quick.orange.rabbit
  • binding key与routing key一样也是句点号.分隔的字符串
  • binding key中可以存在两种特殊的字符*、#，用于做模糊匹配，其中 *用于匹配一个单词，#用于匹配多个单词（可以是零个）。
• headers
  • headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
  • 在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配消息则会路由到该Queue，否则不会路由到该Queue。

1. RPC
   MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）
   但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

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

参考

https://www.cnblogs.com/ericli-ericli/p/5902270.html
https://blog.csdn.net/super_rd/article/details/71191851?utm_source=itdadao&utm_medium=referral