一、RabbitMQ 集群介绍

1.1 两种模式

RabbitMQ 集群有两种模式：
    ●普通集群

    ●镜像集群

1.2 普通集群

普通集群模式，就是将 RabbitMQ 部署到多台服务器上，每台服务器启动一个 RabbitMQ 实例，多个实例之间进行消息通信。

此时我们创建的队列 Queue，它的元数据（主要就是 Queue 的一些配置信息）会在所有的 RabbitMQ 实例中进行同步，但是队列中的消息只会存在于一个 RabbitMQ 实例上，而不会同步到其他队列。

当我们消费消息的时候，如果连接到了另外一个实例，那么那个实例会通过元数据定位到 Queue 所在的位置，然后访问 Queue 所在的实例，拉取数据过来发送给消费者。

这种集群可以提高 RabbitMQ 的消息吞吐能力，但是无法保证高可用，因为一旦一个 RabbitMQ 实例挂了，消息就没法访问了，如果消息队列做了持久化，那么等 RabbitMQ 实例恢复后，就可以继续访问了；如果消息没做持久化，那么消息就丢了。

大致的流程图如下图：

  

1.2 镜像集群

它和普通集群最大的区别在于 Queue 数据和元数据不再是单独存储在一台机器上，而是同时存储在多台机器上。也就是说每个 RabbitMQ 实例都有一份镜像数据（副本数据）。每次写入消息的时候都会自动把数据同步到多台实例上去，这样一旦其中一台机器发生故障，其他机器还有一份副本数据可以继续提供服务，也就实现了高可用。

大致流程图如下图：

  

1.3 节点类型

RabbitMQ 中的节点类型有两种：

    ●RAM node：内存节点将所有的队列、交换机、绑定、用户、权限和 vhost 的元数据定义存储在内存中，好处是可以使得交换机和队列声明等操作速度更快。

    ●Disk node：将元数据存储在磁盘中，单节点系统只允许磁盘类型的节点，防止重启 RabbitMQ 的时候，丢失系统的配置信息。

RabbitMQ 要求在集群中至少有一个磁盘节点，所有其他节点可以是内存节点，当节点加入或者离开集群时，必须要将该变更通知到至少一个磁盘节点。如果集群中唯一的一个磁盘节点崩溃的话，集群仍然可以保持运行，但是无法进行其他操作（增删改查），知道节点恢复。为了确保集群信息的可靠性，或者在不确定使用磁盘节点还是内存节点的时候，建议直接使用磁盘节点。

二、搭建普通集群

2.1 预备知识

    1、搭建集群时，节点中的 Erlang Cookie 值要一致，默认情况下，文件在 /var/lib/rabbitmq/.erlang.cookie，该文件是一个隐藏文件。我们在用 docker 创建 RabbitMQ 容器时，可以位置设置相应的 Cookie 值。
    2、RabbitMQ 是通过主机名来连接服务，必须保证各个主机名之间可以 ping 通。可以通过编辑 /etc/hosts 来手工添加主机名和 IP 对应关系。如果主机名 ping 不通，RabbitMQ 服务启动会失败（如果我们是在不同的服务器上搭建 RabbitMQ 集群，大家需要注意这一点，接下来的 2.2 小结，我们将通过 Docker 容器连接 link 来实现容器之间的访问，略有不同）。

2.2 开始搭建

执行如下命令，创建三个 RabbitMQ 容器

docker run -d --hostname rabbit01 --name mq01 -p 5671:5672 -p 15671:15672 -e RABBITMQ_ERLANG_COOKIE="rabbitmq_cookie" rabbitmq

docker run -d --hostname rabbit02 --name mq02 -p 5672:5672 -p 15672:15672 --link mq01:mylink01 -e RABBITMQ_ERLANG_COOKIE="rabbitmq_cookie" rabbitmq

docker run -d --hostname rabbit03 --name mq03 -p 5673:5672 -p 15673:15672 --link mq01:mylink02 --link mq02:mylink03 -e RABBITMQ_ERLANG_COOKIE="rabbitmq_cookie" rabbitmq

mq02 和 mq03 分别使用了 --link 参数来实现容器连接。注意：mq03 容器既要能够连接 mq01，也要能够连接 mq02。

接下来开始集群的配置

执行如下命令，将 mq02 容器加入到集群中：

docker exec -it mq02 /bin/bash

rabbitmqctl stop_app

rabbitmqctl join_cluster rabbit@rabbit01

rabbitmqctl start_app

执行如下命令，将 mq03 容器加入到集群中：

docker exec -it mq03 /bin/bash

rabbitmqctl stop_app

rabbitmqctl join_cluster rabbit@rabbit01

rabbitmqctl start_app

进入任意一个容器内，输入以下命令查看集群状态：

rabbitmqctl cluster_status

2.3 开启控制台界面

在每一个容器内，执行以下命令：

rabbitmq-plugins enable rabbitmq_management

2.4 控制台界面点击Connection 提示 Stats in management UI are disabled on this node

进入容器内，以 mq01 为例，执行以下命令：

docker exec -it mq01 /bin/bash # 进入容器

cd /etc/rabbitmq/conf.d/ # 进入容器内该目录下

echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf

exit # 退出当前容器

docker restart mq01 # 重启容器

  

2.5 代码测试

配置 application.yml，内容如下：

spring:

    rabbitmq:

        addresses: IP地址:5671,IP地址:5672,IP地址:5673

        username: guest

        password: guest

创建队列、交换机以及绑定关系

@Configuration

public class RabbitConfig {

    public static final String CLUSTER_EXCHANGE_NAME = "cluster_exchange_name";

    public static final String CLUSTER_QUEUE_NAME = "cluster_queue_name";

    public static final String CLUSTER_ROUTING_KEY = "cluster_routing_key";

    @Bean

    public Queue msgQueue(){

        return new Queue(CLUSTER_QUEUE_NAME, true, false, false, null);

    }

    @Bean

    public DirectExchange directExchange(){

        return new DirectExchange(CLUSTER_EXCHANGE_NAME, true, false);

    }

    @Bean

    public Binding queueBindingExchange(@Qualifier("msgQueue") Queue msgQueue,

                                                                  @Qualifier("directExchange") DirectExchange directExchange){

        return BindingBuilder.bind(msgQueue).to(directExchange).with(CLUSTER_ROUTING_KEY);

    }

}

接下来在单元测试中进行消息发送测试：

@SpringBootTest

class RabbitmqClusterApplicationTests {

    @Autowired

    private RabbitTemplate rabbitTemplate;

    @Test

    void contextLoads() {

        rabbitTemplate.convertAndSend(RabbitConfig.CLUSTER_EXCHANGE_NAME, RabbitConfig.CLUSTER_ROUTING_KEY, "你好");

    }

}

消息发送成功中，在 RabbitMQ 的 Web 管理端，会看到三个 RabbitMQ 实例上都会显示一条消息，但是实际上消息本身只存在于一个 RabbitMQ 实例。

接下来创建消息消费者

@Component

public class RabbitConsumer {

    @RabbitListener(queues = RabbitConfig.CLUSTER_QUEUE_NAME)

    public void receiveMsg(Message msg){

        System.out.println("消息：" + new String(msg.getBody()));

    }

}

当消息消费者启动成功后，这个方法中只收到一条消息。

  

2.6 反向测试

确保三个 RabbitMQ 实例都是启动状态，关闭掉 Consumer，然后通过 provider 发送一条消息，发送成功之后，关闭 mq01 实例，然后启动 Consumer 实例，此时 Consumer 实例并不会消费消息，反而会报错说 mq01 实例连接不上，这个例子就可以说明消息在 mq01 上，并没有同步到另外两个 MQ 上。相反，如果 provider 发送消息成功之后，我们没有关闭 mq01 实例而是关闭了 mq02 实例，那么消息的消费并不会受到影响。

使用以下命令停止 mq01 实例：

docker exec -it mq01 /bin/bash

rabbitmqctl stop_app

停止后重新启动 Consumer，会报如下错误

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - home node 'rabbit@rabbit01' of durable queue 'cluster_queue_name' in vhost '/' is down or inaccessible, class-id=50, method-id=10)

    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66) ~[amqp-client-5.14.2.jar:5.14.2]

    at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36) ~[amqp-client-5.14.2.jar:5.14.2]

    at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:502) ~[amqp-client-5.14.2.jar:5.14.2]

    at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:293) ~[amqp-client-5.14.2.jar:5.14.2]

    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141) ~[amqp-client-5.14.2.jar:5.14.2]

    ... 20 common frames omitted

其他节点：

  

三、搭建镜像集群

镜像集群不需要额外搭建，只需要将队列配置为镜像队列即可。这个配置可以通过网页配置，也可以通过命令行配置。

3.1 网页配置镜像队列

点击 Admin 选项卡，然后点击右边的 Policies，再点击 Add/update a policy，如下图：

  
接下来，添加一个策略，如下图：

  
各参数含义如下：

    ●Name：policy 的名称

    ●Pattern：queue 的匹配模式（正则表达式）

    ●Definition：镜像定义，主要由三个参数：ha-mode，ha-params，ha-sync-mode。

    - ha-mode：指明镜像队列的模式，有效值为 all、exactly、nodes。其中 all 表示在集群中所有的节点上进行镜像（默认即此）；exactly 表示在指定个数的节点上进行镜像，节点的个数由 ha-params 指定；nodes 表示在指定的节点上进行镜像，节点名称通过 ha-params 指定。
    - ha-params：ha-mode 模式需要用到的参数
    - ha-sync-mode：进行队列中消息的同步方式，有效值为 automatic 和 manual。
    ●priority 为可选参数，表示 policy 的优先级

配置完成后，点击下面的 add/update policy 按钮，完成策略的添加，如下：

  
添加完成后，进行一个简单测试：
首先确认三个 RabbitMQ 都启动了，然后用上面的 provider 向消息队列发送一条消息。发送完成之后关闭 mq01 实例。接下来启动 consumer，此时发现 consumer 可以完成消息的消费（注意和前面的反向测试区分），这就说明镜像队列已经搭建成功了。

3.2 命令行配置镜像队列

命令行的配置格式如下：

rabbitmqctl set_policy [-p vhost] [--priority priority] [--apply-to apply-to] {name} {pattern} {definition}

举一个简单的配置案例：

rabbitmqctl set_policy -p / --apply-to queues my_queue_mirror "^" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

免责申明：

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！

《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack