【04】RabbitMQ的集群机制

1、RabbitMQ的性能监控

• 关于RabbitMQ的性能监控，在管理控制台中提供了非常丰富的展示。例如：首页这个整体监控页面，就展示了非常多详细的信息：
• 还包括消息的生产消费频率、关键组件的使用情况等等非常多的消息。都可以在这个管理控制台上展现出来。但是，对于构建一个自动化的性能监控系统来说，这个管理页面是不够用的。为此，RabbitMQ也提供了一系列的HTTP接口，通过这些接口可以非常全面的监控并且管理RabbitMQ的各种功能。
• 这些HTTP的接口不需要专门去查手册，在部署的管理控制台页面下方已经集成了详细的文档，我们只需要打开HTTP API的页面就能看到。
• 比如最常用的 http://[server:port]/api/overview 接口，会列出非常多的信息，包含系统的资源使用情况。通过这个接口，就可以很好的对接Promethus、Grafana等工具，构建更灵活的监控告警体系。
• 可以看到，这里面的接口相当丰富，不光可以通过GET请求获取各种消息，还可以通过其他类型的HTTP请求来管理RabbitMQ中的各种资源，因此在实际使用时，还需要考虑这些接口的安全性。

2、RabbitMQ的备份与恢复

• RabbitMQ有一个data目录，会保存分配到该节点上的所有消息。默认目录存放路径为：/var/lib/rabbitmq/mnesia，这个目录中的备份分为两个部分，一个是元数据（定义结构的数据），另一个是消息存储目录。
• 对于元数据，可以在Web管理页面通过json文件直接导出或者导入：
• 对于消息，可以手动进行备份恢复：- 其实对于消息，由于MQ的特性，是不建议进行备份恢复的。而RabbitMQ如果要进行数据备份恢复，也非常简单。- 首先，要保证要恢复的RabbitMQ中已经有了全部的元数据，这个可以通过上一步的json文件来恢复。- 然后，备份过程必须要先停止应用。如果是针对镜像集群，还需要把整个集群全部停止。- 最后，在RabbitMQ的数据目录中，有按virtual hosts组织的文件夹。你只需要按照虚拟主机，将整个文件夹复制到新的服务中即可。持久化消息和非持久化消息都会一起备份。- 最后注意：拷贝后的文件权限需要改为rabbitmq用户，使用命令：chown -R rabbitmq:rabbitmq ./*

3、使用联邦插件进行远程消息同步

3.1 插件的作用

• 在企业中有很多大型的分布式场景，在这些业务场景下，希望服务也能够同样进行分布式部署。这样既可以提高数据的安全性，也能够提升消息读取的性能。例如，某大型企业，可能在北京机房和西安机房分别搭建RabbitMQ服务，然后希望西安机房需要同步北京机房的消息，这样可以让西安的消费者服务可以直接连接西安本地的RabbitMQ，而不用费尽周折去连接北京机房的RabbitMQ服务。这时要如何进行数据同步呢？搭建一个跨度这么大的内部子网显然就不太划算。这时就可以考虑使用RabbitMQ的Federation插件，搭建联邦队列Federation。通过Federation可以搭建一个单向的数据同步通道。

3.2 使用步骤

• <1> 启动插件- RabbitMQ的官方运行包中已经包含了Federation插件。只需要启动后就可以直接使用。- 相关命令：# 确认联邦插件rabbitmq-plugins list|grep federation[] rabbitmq_federation 3.13.6[] rabbitmq_federation_management 3.13.6# 启用联邦插件rabbitmq-plugins enable rabbitmq_federation# 启用联邦插件的管理平台支持rabbitmq-plugins enable rabbitmq_federation_management- 插件启用完成后，可以在管理控制台的Admin菜单看到两个新增选项 Federation Status和Federation Upstreams。
• <2> 配置Upstream- Upstream表示是一个外部的服务节点，在RabbitMQ中，可以是一个交换机，也可以是一个队列。它的配置方式是由下游服务主动配置一个与上游服务的连接，然后数据就会从上游服务主动同步到下游服务中。- 接下来我们用192.168.65.112上的的RabbitMQ服务来模拟DownStream下游服务，去指向一个192.168.65.193服务器上搭建的RabbitMQ服务，搭建一个联邦交换机Federation Exchange。 - 首先要在下游RabbitMQ中声明一个交换机和交换队列，用来接收远端的数据。这里我们直接用客户端API来快速进行声明。publicclassDownStreamConsumer{publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{ConnectionFactory factory =newConnectionFactory(); factory.setHost("192.168.65.112"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); factory.setVirtualHost("/mirror");Connection connection = factory.newConnection();Channel channel = connection.createChannel(); channel.exchangeDeclare("fed_exchange","direct"); channel.queueDeclare("fed_queue",true,false,false,null); channel.queueBind("fed_queue","fed_exchange","routKey");Consumer myconsumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println("========================");String routingKey = envelope.getRoutingKey();System.out.println("routingKey >"+ routingKey);String contentType = properties.getContentType();System.out.println("contentType >"+ contentType);long deliveryTag = envelope.getDeliveryTag();System.out.println("deliveryTag >"+ deliveryTag);System.out.println("content:"+newString(body,"UTF-8"));}}; channel.basicConsume("fed_queue",true, myconsumer);}}- 然后在下游RabbitMQ服务中配置一个上游服务。- 服务的名字Name属性随意，URI指向远程服务器(配置方式参看页面上的示例)：amqp://admin:[email protected]:5672/，需要注意下，下游和上游建议使用相同的虚拟机。- 下面的Federated exchanges parameters和Federated queues parameters分别指定 上游(也就是远程服务器)的Exchange和Queue。如果不指定，就是用和下游中相同的Exchange和Queue。如果UpStream里没有，就创建新的Exchange和Queue。
• <3> 配置 Federation 策略：- 接下来需要配置一个指向上游服务的Federation策略。在配置策略时可以选择是针对Exchange交换机还是针对Queue队列。配置策略时，同样有很多参数可以选择配置。最简化的一个配置如下：- 注意：每个策略的Definition部分，至少需要指定一个Federation目标。federation-upstream-set参数表示是以set集合的方式针对多个Upstream生效，all表示是全部Upstream。而federation-upstream参数表示只对某一个Upstream生效。
• <4> 测试：- 配置完Upstream和对应的策略后，进入Federation Status菜单就能看到Federation插件的执行情况。状态为running表示启动成功，如果配置出错，则会提示失败原因。- 然后，在远程服务193的RabbitMQ服务中，可以看到对应生成的Federation交换机。- 并且，在fed_exchange的详情页中也能够看到绑定关系。这里要注意一下它给出了一个默认的Routing_key。- 接下来就可以尝试在上游服务193的fed_exchange中发送消息，消息会同步到Local本地的联邦交换机中，从而被对应的消费者消费到。

4、RabbitMQ服务高可用机制

4.1 RabbitMQ的集群机制

• 在企业中真实使用RabbitMQ时，单机肯定是不够的。如果只是单机RabbitMQ的服务崩溃了，那还好，大不了重启下服务就是了。但是如果是服务器的磁盘出问题了，那问题就大了。因为消息都是存储在Queue里的，Queue坏了，意味着消息就丢失了。这在生产环境上肯定是无法接受的。而RabbitMQ的设计重点就是要保护消息的安全性。
• 所以RabbitMQ在设计之初其实就采用了集群模式来保护消息的安全。基础的思想就是给每个Queue提供几个备份。当某一个服务的Queue坏了，至少还可以从其他Queue中获取服务。
• 其实对于RabbitMQ，一个节点的服务也是作为一个集群来处理的，在web控制台的admin-> cluster 中可以看到集群的名字，并且可以在页面上修改。
• 实际上，RabbitMQ实现了两种集群模式： - 普通集群模式（默认）：- 这种模式使用Erlang语言天生具备的集群方式搭建。这种集群模式下，集群的各个节点之间只会有相同的元数据，即队列结构，而消息不会进行冗余，只存在一个节点中。消费时，如果消费的不是存有数据的节点， RabbitMQ会临时在节点之间进行数据传输，将消息从存有数据的节点传输到消费的节点。- 很显然，这种集群模式的消息可靠性不是很高。因为如果其中有个节点服务宕机了，那这个节点上的数据就无法消费了，需要等到这个节点服务恢复后才能消费，而这时，消费者端已经消费过的消息就有可能给不了服务端正确应答，服务起来后，就会再次消费这些消息，造成这部分消息重复消费。 另外，如果消息没有做持久化，重启就消息就会丢失。- 并且，这种集群模式也不支持高可用，即当某一个节点服务挂了后，需要手动重启服务，才能保证这一部分消息能正常消费。- 所以这种集群模式只适合一些对消息安全性不是很高的场景。而在使用这种模式时，消费者应该尽量的连接上每一个节点，减少消息在集群中的传输。- 镜像集群模式：- 这种模式是在普通集群模式基础上的一种增强方案，这也就是RabbitMQ的官方HA高可用方案。需要在搭建了普通集群之后再补充搭建。其本质区别在于，这种模式会在镜像节点中间主动进行消息同步，而不是在客户端拉取消息时临时同步。- 并且在集群内部有一个算法会选举产生master和slave，当一个master挂了后，也会自动选出一个来。从而给整个集群提供高可用能力。- 这种模式的消息可靠性更高，因为每个节点上都存着全量的消息。而他的弊端也是明显的，集群内部的网络带宽会被这种同步通讯大量的消耗，进而降低整个集群的性能。这种模式下，队列数量最好不要过多。

4.2 搭建普通集群

• 我们准备三台服务器，在三台服务器上分别搭建起RabbitMQ的服务。在实际项目中，通常建议搭建奇数台服务的集群，因为这样的集群对官方推荐的Quorum对列更友好。
• 然后需要再三天服务器上调整域名映射：# 配置每台机器的域名 vi /etc/hosts 192.168.65.193 192-168-65-193 192.168.65.112 192-168-65-112 192.168.65.170 192-168-65-170
• 接下来还需要将各个节点上的集群名字调整为 rabbit@worker1这样的形式。每个节点的集群名字对应自己的域名。做好这些准备工作，就可以来搭建集群了。- 1、需要同步集群节点中的cookie- 默认会在 /var/lib/rabbitmq/目录下生成一个.erlang.cookie。 里面有一个字符串。我们要做的就是保证集群中三个节点的这个cookie字符串一致。需要使用 ll -a 才能看到这个隐藏文件。- 我们将worker2和3加入到worker1的RabbitMQ集群中，所以将worker1的.erlang.cookie文件分发到worker2和3，使用命令 scp .erlang.cookie [email protected]:/var/lib/rabbitmq。- 同步文件时注意一下文件的权限，如果文件不可读，集群启动会有问题。需要将文件的所属用户调整为rabbitmq。指令为 chown rabbitmq:rabbitmq .erlang.cookie。 另外需要注意文件的权限，必须只有当前用户可读。建议使用指令 chmod 400 .erlang.cookie 重新调整下文件权限。- 2、将worker2和3的服务加入到worker1的集群中- 首先需要保证worker1上的rabbitmq服务是正常启动的。 然后分别在worker2和3上执行以下指令： - RabbitMQ的集群节点分为disk和ram。disk节点会将元数据保存到硬盘当中，而ram节点是将元数据保存到内存当中的。- 由于ram节点减少了很多与硬盘的交互，所以，ram节点的元数据使用性能会比较高。但是，这也意味着元数据的安全性是不如disk节点的。在我们这个集群中，worker2和worker3都以ram节点的身份加入到worker1集群里，因此，是存在单点故障的。如果worker1节点服务崩溃，那么元数据就有可能丢失。在企业进行部署时，性能与安全性需要自己进行平衡。- 这里说的元数据仅仅只包含交换机、队列等的定义，而不包含具体的消息。因此，ram节点的性能提升，仅仅体现在对元数据进行管理时，比如修改队列queue，交换机exchange，虚拟机vhosts等时，与消息的生产和消费速度无关。- 如果一个集群中，全部都是ram节点，那么元数据就有可能丢失。这会造成集群停止之后就启动不起来了。RabbitMQ会尽量阻止创建一个全是ram节点的集群，但是并不能彻底阻止。所以，综合考虑，官方其实并不建议使用ram节点，更推荐保证集群中节点的资源投入，使用disk节点。
• 3、加入完成之后，可以查看集群的状态- 使用指令查看集群状态：rabbitmqctl cluster_status- 在Web管理界面查看集群状态：

4.3 搭建镜像集群

• 在普通集群的基础之上，可以继续搭建镜像集群。
• 通常在生产环境中，为了减少RabbitMQ集群之间的数据传输，在配置镜像策略时，会针对固定的虚拟主机virtual host来配置。
• RabbitMQ中的 vritual host 可以类比为MySQL中的库，针对每个虚拟主机，可以配置不同的权限、策略等。并且不同虚拟主机之间的数据是相互隔离的。
• 我们首先创建一个/mirror的虚拟主机，然后再添加给对应的镜像策略：# 创建虚拟机rabbitmqctl add_vhost /mirror# 添加镜像策略rabbitmqctl set_policy ha-all --vhost "/mirror""^"'{"ha-mode":"all"}'
• 这些参数需要大致了解下。其中，pattern 是队列的匹配规则, ^ 表示全部匹配，^ ha \ 这样的配置表示以ha开头。通常就用虚拟主机来区分就够了，这个队列匹配规则就配置成全匹配。
• 然后几个关键的参数：- HA mode: 可选值 all , exactly, nodes。生产上通常为了保证高可用，就配all- all : 队列镜像到集群中的所有节点。当新节点加入集群时，队列也会被镜像到这个节点。- exactly : 需要搭配一个数字类型的参数(ha-params)。队列镜像到集群中指定数量的节点。如果集群内节点数少于这个数字，则队列镜像到集群内的所有节点。如果集群内节点少于这个数，当一个包含镜像的节点停止服务后，新的镜像就不会去另外找节点进行镜像备份了。- nodes: 需要搭配一个字符串类型的参数。将队列镜像到指定的节点上。如果指定的队列不在集群中，不会报错。当声明队列时，如果指定的所有镜像节点都不在线，那队列会被删除。
• 通常镜像模式的集群已经足够满足大部分的生产场景了。虽然他对系统资源消耗比较高，但是在生产环境中，系统的资源都是会做预留的，所以正常的使用是没有问题的。但是在做业务集成时，还是需要注意队列数量不宜过多，并且尽量不要让RabbitMQ产生大量的消息堆积。