RabbitMQ从零到集群高可用(.NetCore5.0) -高可用集群构建落地

2021-12-31 00:53:23

系列文章：

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 ，两台主机上的搭建的步骤完全相同，只是部分配置略有不同，具体如下：

官网：https://www.keepalived.org

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();
        }