生产者代码：

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