- A+
系列文章：
RabbitMQ从零到集群高可用(.NetCore5.0) - RabbitMQ简介和六种工作模式详解
RabbitMQ从零到集群高可用(.NetCore5.0) - 死信队列，延时队列
RabbitMQ从零到集群高可用(.NetCore5.0) - 高可用集群构建落地
一、集群架构简介
当单台 RabbitMQ 服务器的处理消息的能力达到瓶颈时，此时可以通过 RabbitMQ 集群来进行扩展，从而达到提升吞吐量的目的。RabbitMQ 集群是一个或多个节点的逻辑分组，集群中的每个节点都是对等的，每个节点共享所有的用户，虚拟主机，队列，交换器，绑定关系，运行时参数和其他分布式状态等信息。一个高可用，负载均衡的 RabbitMQ 集群架构应类似下图：
解析说明：
最下面层是RabbitMQ的集群，没有ha镜像时是普通集群，普通集群的缺点是挂了一个机器，以这个机器为根的队列就无法使用了（一个队列的数据只会存在一个节点），无法实现高可用。
所以把队列变成镜像队列，这样每个节点都会有一份完整的数据，有节点挂了也不影响使用，实现了高可用，但RabbitMQ集群本身没有实现负载均衡，也就是说对于一个三节点的集群，
每个节点的负载可能都是不相同的。
HAProxy层的作用就是为了实现RabbitMQ集群的负载均衡，但一个节点的话显然也不能高可用，所以需要两个HAProxy实现HaProxy的高可用，但没法实现自动的故障转移，就是HAProxy1挂了，
需要手动把ip地址改成HAProxy2的。
所以需要用到KeepAlived，它通常由一主一备两个节点组成，同一时间内只有主节点会提供对外服务，并同时提供一个虚拟的 IP 地址 (Virtual Internet Protocol Address ，简称 VIP)，可以避免暴露真实ip 。 如果主节点故障，那么备份节点会自动接管 VIP 并成为新的主节点 ，直到原有的主节点恢复。
生产环境架构应该为：
机器1：RabbitMQ1，机器2：RabbitMQ2，机器3：RabbitMQ3，机器4：HAProxy+keeplived(主)，机器5(HAProxy+keeplived(备)。
这里资源原因只有3台机器，所以搭建架构为：
| 172.16.2.84(rabbit1) | RabbitMQ1，HAProxy+keeplived(主) |
| 172.16.2.85(rabbit2) | RabbitMQ2，HAProxy+keeplived(备) |
| 172.16.2.86(rabbit3) | RabbitMQ3 |
二、普通集群搭建
1）各个节点分别安装RabbitMQ
这里前面的章节功能使用是用的docker安装，这里集群不用docker，因为docker安装会有很多的映射，很容易扰乱，装有docker的rabbitmq的，需要先把docker停掉。
这里准备了3台机器，172.16.2.84（rabbit1），172.16.2.85(rabbit2)，172.16.2.86(rabbit3)，为了避免混淆，下面都会用rabbit1，rabbit2，rabbit3称呼。
rabbit1，rabbit2，rabbit3都执行一遍下面的安装。
1.1安装前，先修改hostname，因为rabbitmq集群通讯需要用hostname
rabbit1机器执行
hostnamectl set-hostname rabbit1 --static
rabbit2机器执行
hostnamectl set-hostname rabbit2 --static
rabbit3机器执行
hostnamectl set-hostname rabbit3 --static
rabbit1，rabbit2，rabbit3执行同样操作
#修改hosts，因为rabbitmq通讯要通过hostname vi /etc/hosts
1.2把ip对应hostname添加到后面
172.16.2.84 rabbit1
172.16.2.85 rabbit2
172.16.2.86 rabbit3
# 重启网络 systemctl restart network # 重启机器 init 6
1.3安装Erlang
#第一步 运行package cloud提供的erlang安装脚本 curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash #第二步 安装erlang yum install erlang #第三步 查看erlang版本号,在命令行直接输入erl erl
1.4安装RabbitMQ
#第一步 先导入两个key rpm --import https://packagecloud.io/rabbitmq/rabbitmq-server/gpgkey rpm --import https://packagecloud.io/gpg.key #第二步 运行package cloud提供的rabbitmq安装脚本 curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash #第三步 下载rabbit安装文件 wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.9.5/rabbitmq-server-3.9.5-1.el8.noarch.rpm #第四步 rpm --import https://www.rabbitmq.com/rabbitmq-release-signing-key.asc #第五步 rabbitMQ依赖 yum -y install epel-release yum -y install socat #第六步 安装 rpm -ivh rabbitmq-server-3.9.5-1.el8.noarch.rpm #第七步 启用管理平台插件，启用插件后，可以可视化管理RabbitMQ rabbitmq-plugins enable rabbitmq_management #第八步 启动应用 systemctl start rabbitmq-server
上面rabbitmq的版本号安装文件地址：https://github.com/rabbitmq/rabbitmq-server/releases/
1.5设置访问权限
#创建管理员账户 rabbitmqctl add_user admin 123456 #设置注册的账户为管理员 rabbitmqctl set_user_tags admin administrator #授权远程访问 rabbitmqctl set_permissions -p / admin "." "." ".*" #重启服务 systemctl restart rabbitmq-server
注意这里关了防火墙，如果开启防火墙的话，需要把15672端口开放出来
#查看防火墙状态 systemctl status firewalld #关闭防火墙 systemctl stop firewalld
到这里，3台机器都安装好了RabbitMQ。
2）把节点加入集群
2.1停止服务rabbit2,rabbit3
#停止全部服务 systemctl stop rabbitmq-server
2.2拷贝cookie
将rabbit1上的.erlang.cookie文件拷贝到其他两台主机上。该 cookie 文件相当于密钥令牌，集群中的 RabbitMQ 节点需要通过交换密钥令牌以获得相互认证，因此处于同一集群的所有节点需要具有相同的密钥令牌，否则在搭建过程中会出现 Authentication Fail 错误，只要保证这3台机器中的.erlang.cookie内的密钥字符串一致即可，这里把rabbit1的密钥复制到rabbit2，rabbit3。
3个机器都给.erlang.cookie 400权限。
#给600权限 chmod 600 /var/lib/rabbitmq/.erlang.cookie
rabbit1机器
#编辑文件 vi /var/lib/rabbitmq/.erlang.cookie
把内容复制出来，修改rabbit2，rabbit3这个文件的值为这个。
启动服务
#开启全部服务 systemctl start rabbitmq-server
2.3集群搭建
RabbitMQ 集群的搭建需要选择其中任意一个节点为基准，将其它节点逐步加入。这里我们以 rabbit1为基准节点，将 rabbit2 和 rabbit3 加入集群。在 rabbit2和 rabbit3上执行以下命令：
# 1.停止服务 rabbitmqctl stop_app # 2.重置状态（需要更改节点类型的时候执行，首次不需要执行，除非你节点是以disk加入集群的） rabbitmqctl reset # 3.节点加入 #rabbitmqctl join_cluster --ram rabbit@rabbit1 rabbitmqctl join_cluster rabbit@rabbit1 # 4.启动服务 rabbitmqctl start_app
join_cluster 命令有一个可选的参数 --ram ，该参数代表新加入的节点是内存节点，默认是磁盘节点。如果是内存节点，则所有的队列、交换器、绑定关系、用户、访问权限和 vhost 的元数据都将存储在内存中，如果是磁盘节点，则存储在磁盘中。内存节点可以有更高的性能，但其重启后所有配置信息都会丢失，因此RabbitMQ 要求在集群中至少有一个磁盘节点，其他节点可以是内存节点，大多数情况下RabbitMQ 的性能都是够用的，可以采用默认的磁盘节点的形式。
另外，如果节点以磁盘节点的形式加入，则需要先使用 reset 命令进行重置，然后才能加入现有群集，重置节点会删除该节点上存在的所有的历史资源和数据。采用内存节点的形式加入时可以略过 reset 这一步，因为内存上的数据本身就不是持久化的。
操作上面的，一个普通集群就搭建成功了，打开rabbit管理界面(随便打开一个都是一样的)。
3）代码演示普通集群的问题
普通集群中， 第一次创建队列时，会随机选一个节点作为根节点，这个节点会存储队列的信息(交互机，路由，队列名等)和队列的数据，其它两个节点只会同步根节点的元信息(交换机，路由，队列名)等，
但不会存储队列的数据，他们是通过元信息找到根节点读写消息。
例如集群选择了rabbit2作为根节点，那么数据存储在rabbit2，rabbit1和rabbit3是没有数据的，那么如果rabbit2宕机了，队列里面的数据就取不到了。
代码演示，.NetCore5.0读取集群连接代码。
/// <summary> /// 获取集群连接对象 /// </summary> /// <returns></returns> public static IConnection GetClusterConnection() { var factory = new ConnectionFactory { UserName = "admin",//账户 Password = "123456",//密码 VirtualHost = "/" //虚拟机 }; List<AmqpTcpEndpoint> list = new List<AmqpTcpEndpoint>() { new AmqpTcpEndpoint(){HostName="172.16.2.84",Port=5672}, new AmqpTcpEndpoint(){HostName="172.16.2.85",Port=5672}, new AmqpTcpEndpoint(){HostName="172.16.2.86",Port=5672} }; return factory.CreateConnection(list); }
生产者代码：
/// <summary> /// 工作队列模式 /// </summary> public static void WorkerSendMsg() { string queueName = "worker_order";//队列名 //创建连接 using (var connection = RabbitMQHelper.GetClusterConnection()) { //创建信道 using (var channel = connection.CreateModel()) { //创建队列 channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); IBasicProperties properties = channel.CreateBasicProperties(); properties.Persistent = true; //消息持久化 for ( var i=0;i<10;i++) { string message = $"Hello RabbitMQ MessageHello,{i+1}"; var body = Encoding.UTF8.GetBytes(message); //发送消息到rabbitmq channel.BasicPublish(exchange: "", routingKey: queueName, mandatory: false, basicProperties: properties, body); Console.WriteLine($"发送消息到队列:{queueName},内容:{message}"); } } } }
执行：
查看RabbitMQ管理界面
RabbitMQ选择了rabbit3作为根节点，现在试一下停止rabbit3节点
#停止服务 rabbitmqctl stop_app
发现队列不可用了，启动rabbit3，再试一下停止rabbit2
发现队列正常，试下消费数据：
public static void WorkerConsumer() { string queueName = "worker_order"; var connection = RabbitMQHelper.GetClusterConnection(); { //创建信道 var channel = connection.CreateModel(); { //创建队列 channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); var consumer = new EventingBasicConsumer(channel); ///prefetchCount:1来告知RabbitMQ,不要同时给一个消费者推送多于 N 个消息，也确保了消费速度和性能 ///global：是否设为全局的 ///prefetchSize:单条消息大小，通常设0，表示不做限制 //是autoAck=false才会有效 channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true); int i = 1; int index = new Random().Next(10); consumer.Received += (model, ea) => { //处理业务 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{i},消费者:{index},队列{queueName}消费消息长度:{message.Length}"); Thread.Sleep(1000); channel.BasicAck(ea.DeliveryTag, false); //消息ack确认，告诉mq这条队列处理完，可以从mq删除了 i++; }; channel.BasicConsume(queueName, autoAck: false, consumer); } } }
如果根节点挂了，再往集群发送数据，RabbitMQ又会从其余的选一个作为根节点，就可能的情况是，多个节点都存有不同队列的数据。
到这里，可以看到普通集群并不可以高可用， 根节点挂了，在这个节点上的队列也不可用了。但三个节点都没问题的时候，可以提高并发和队列的负载。
要实现高可用，就要用到了镜像集群。
三、镜像集群
镜像集群，在每个节点上都同步一份镜像数据，相当于每个节点都有一份完整的数据，这样有节点宕机了，还能正常提供RabbitMQ服务。
变镜像集群很简单，在上面普通集群的基础上，在任意一个节点下执行
#把集群变成镜像集群 rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
执行完，看回队列，多了ha标识。
四、HAProxy环境搭建。
4.1下载
HAProxy 官方下载地址为：https://www.haproxy.org/#down ，如果这个网站无法访问，也可以从 https://src.fedoraproject.org/repo/pkgs/haproxy/ 上进行下载。
#下载haproxy wget https://www.haproxy.org/download/2.4/src/haproxy-2.4.3.tar.gz
解压
#解压 tar xf haproxy-2.4.3.tar.gz
4.2编译
安装gcc
yum install gcc
进入解压后根目录，执行下面的编译命令：
#进入haproxy-2.4.3目录后执行 make TARGET=linux-glibc PREFIX=/usr/app/haproxy-2.4.3 make install PREFIX=/usr/app/haproxy-2.4.3
4.3配置环境变量
#编辑文件 vi /etc/profile
#把这两项加上 export HAPROXY_HOME=/usr/app/haproxy-2.4.3 export PATH=$PATH:$HAPROXY_HOME/sbin
使得配置的环境变量立即生效：
source /etc/profile
4.4负载均衡配置
新建配置文件haproxy.cfg，这里我新建的位置为：/etc/haproxy/haproxy.cfg，文件内容如下：
global # 日志输出配置、所有日志都记录在本机，通过 local0 进行输出 log 127.0.0.1 local0 info # 最大连接数 maxconn 4096 # 改变当前的工作目录 chroot /usr/app/haproxy-2.4.3 # 以指定的 UID 运行 haproxy 进程 uid 99 # 以指定的 GID 运行 haproxy 进程 gid 99 # 以守护进行的方式运行 daemon # 当前进程的 pid 文件存放位置 pidfile /usr/app/haproxy-2.4.3/haproxy.pid # 默认配置 defaults # 应用全局的日志配置 log global # 使用4层代理模式，7层代理模式则为"http" mode tcp # 日志类别 option tcplog # 不记录健康检查的日志信息 option dontlognull # 3次失败则认为服务不可用 retries 3 # 每个进程可用的最大连接数 maxconn 2000 # 连接超时 timeout connect 5s # 客户端超时 timeout client 120s # 服务端超时 timeout server 120s # 绑定配置 listen rabbitmq_cluster bind :5671 # 配置TCP模式 mode tcp # 采用加权轮询的机制进行负载均衡 balance roundrobin # RabbitMQ 集群节点配置 server rabbit1 rabbit1:5672 check inter 5000 rise 2 fall 3 weight 1 server rabbit2 rabbit2:5672 check inter 5000 rise 2 fall 3 weight 1 server rabbit3 rabbit3:5672 check inter 5000 rise 2 fall 3 weight 1 # 配置监控页面 listen monitor bind :8100 mode http option httplog stats enable stats uri /stats stats refresh 5s
上传上去的，记得打开看一下换行符有没有出行这些符号，如果有要删掉，不然会报错。
负载均衡的主要配置在 listen rabbitmq_cluster 下，这里指定负载均衡的方式为加权轮询，同时定义好健康检查机制：
server rabbit1 rabbit1:5672 check inter 5000 rise 2 fall 3 weight 1
以上配置代表对地址为 rabbit1:5672 的 rabbit1 节点每隔 5 秒进行一次健康检查，如果连续两次的检查结果都是正常，则认为该节点可用，此时可以将客户端的请求轮询到该节点上；如果连续 3 次的检查结果都不正常，则认为该节点不可用。weight 用于指定节点在轮询过程中的权重。
4.5启动服务
haproxy -f /etc/haproxy/haproxy.cfg
启动后可以在监控页面进行查看，端口为设置的 8100，完整地址为：http://172.16.2.84:8100/stats ，页面情况如下：
所有节点都为绿色，代表节点健康。此时证明 HAProxy 搭建成功，并已经对 RabbitMQ 集群进行监控。
这里已经实现了RabbitMQ的负载均衡了，代码怎么通过Haproxy连接Rabbit集群呢，因为上面配置Haproxy暴露的端口是5671，所以Ip是Haproxy的ip:5671。
连接代码，可以通过haproxy1 或haproxy2都可以：
public static IConnection GetConnection() { ConnectionFactory factory = new ConnectionFactory { HostName = "172.16.2.84",//haproxy ip Port = 5671,//haproxy 端口 UserName = "admin",//账号 Password = "123456",//密码 VirtualHost = "/" //虚拟主机 }; return factory.CreateConnection(); }
五、KeepAlived 环境搭建
接着就可以搭建 Keepalived 来解决 HAProxy 故障转移的问题。这里我在 rabbit1 和 rabbit2 上安装 KeepAlived ，两台主机上的搭建的步骤完全相同，只是部分配置略有不同，具体如下：
5.1安装
yum install -y keepalived
5.2修改配置文件
安装了keepalived后，配置文件生成在/etc/keepalived/keepalived.conf
这里先对keepalived1上keepalived.conf配置文件进行修改，完整内容如下：
global_defs { # 路由id,主备节点不能相同 router_id node1 } # 自定义监控脚本 vrrp_script chk_haproxy { # 脚本位置 script "/etc/keepalived/haproxy_check.sh" # 脚本执行的时间间隔 interval 5 weight 10 } vrrp_instance VI_1 { # Keepalived的角色，MASTER 表示主节点，BACKUP 表示备份节点 state MASTER # 指定监测的网卡，可以使用 ip addr 进行查看 interface ens33 # 虚拟路由的id，主备节点需要设置为相同 virtual_router_id 1 # 优先级，主节点的优先级需要设置比备份节点高 priority 100 # 设置主备之间的检查时间，单位为秒 advert_int 1 # 定义验证类型和密码 authentication { auth_type PASS auth_pass 123456 } # 调用上面自定义的监控脚本 track_script { chk_haproxy } virtual_ipaddress { # 虚拟IP地址，可以设置多个 172.16.2.200 } }
以上配置定义了 keepalived1上的 Keepalived 节点为 MASTER 节点，并设置对外提供服务的虚拟 IP 为 172.16.2.200。此外最主要的是定义了通过 haproxy_check.sh 来对 HAProxy 进行监控，这个脚本需要我们自行创建，内容如下：
#!/bin/bash # 判断haproxy是否已经启动 if [ `ps -C haproxy --no-header | wc -l` -eq 0 ] ; then #如果没有启动，则启动 haproxy -f /etc/haproxy/haproxy.cfg fi #睡眠3秒以便haproxy完全启动 sleep 3 #如果haproxy还是没有启动，此时需要将本机的keepalived服务停掉，以便让VIP自动漂移到另外一台haproxy if [ `ps -C haproxy --no-header | wc -l` -eq 0 ]; then systemctl stop keepalived fi
创建后为其赋予执行权限：
chmod +x /etc/keepalived/haproxy_check.sh
这个脚本主要用于判断 HAProxy 服务是否正常，如果不正常且无法启动，此时就需要将本机 Keepalived 关闭，从而让虚拟 IP 漂移到备份节点。
备份节点(keepalived2)的配置与主节点基本相同，但是需要修改其 state 为 BACKUP；同时其优先级 priority 需要比主节点低。完整配置如下：
global_defs { # 路由id,主备节点不能相同 router_id node2 } vrrp_script chk_haproxy { script "/etc/keepalived/haproxy_check.sh" interval 5 weight 10 } vrrp_instance VI_1 { # BACKUP 表示备份节点 state BACKUP interface ens33 virtual_router_id 1 # 优先级，备份节点要比主节点低 priority 50 advert_int 1 authentication { auth_type PASS auth_pass 123456 } track_script { chk_haproxy } virtual_ipaddress { 172.16.2.200 } }
haproxy_check.sh文件和keepalived1相同
5.3启动服务
分别在KeepAlived1和KeepAlived2上启动KeepAlived服务，命令如下：
systemctl start keepalived
启动后此时 keepAlived1 为主节点，可以在 keepAlived1 上使用 ip a 命令查看到虚拟 IP 的情况：
此时只有 keepAlived1 上是存在虚拟 IP 的，而 keepAlived2 上是没有的。
5.4验证故障转移
这里我们验证一下故障转移，因为按照我们上面的检测脚本，如果 HAProxy 已经停止且无法重启时 KeepAlived 服务就会停止，这里我们直接使用以下命令停止 Keepalived1 服务：
systemctl stop keepalived
此时再次使用 ip a 分别查看，可以发现 keepalived1上的 VIP 已经漂移到 keepalived2上，情况如下：
此时对外服务的 VIP 依然可用，代表已经成功地进行了故障转移。至此集群已经搭建成功，任何需要发送或者接受消息的客户端服务只需要连接到该 VIP 即可，示例如下：
public static IConnection GetConnection() { ConnectionFactory factory = new ConnectionFactory() { HostName = "172.16.2.200",//vip Port = 5671,//haproxy 端口 UserName = "admin",//账号 Password = "123456",//密码 VirtualHost = "/" //虚拟主机 }; return factory.CreateConnection(); }
