运用.net core中实例讲解RabbitMQ高可用集群构建

# 重启网络
systemctl restart network
# 重启机器
init 6

#创建管理员账户
rabbitmqctl add_user admin 123456
#设置注册的账户为管理员
rabbitmqctl set_user_tags admin administrator
#授权远程访问
rabbitmqctl set_permissions -p / admin "." "." ".*"
#重启服务
systemctl restart rabbitmq-server

#查看防火墙状态
systemctl status firewalld
#关闭防火墙
systemctl stop firewalld

#停止全部服务
systemctl stop rabbitmq-server

#给600权限
chmod 600 /var/lib/rabbitmq/.erlang.cookie

#开启全部服务
systemctl start rabbitmq-server

# 1.停止服务
rabbitmqctl stop_app
# 2.重置状态（需要更改节点类型的时候执行，首次不需要执行，除非你节点是以disk加入集群的）
rabbitmqctl reset
# 3.节点加入
#rabbitmqctl join_cluster --ram rabbit@rabbit1
rabbitmqctl join_cluster rabbit@rabbit1
# 4.启动服务
rabbitmqctl start_app

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

#停止服务
rabbitmqctl stop_app

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

#把集群变成镜像集群
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

#下载haproxy
wget https://www.haproxy.org/download/2.4/src/haproxy-2.4.3.tar.gz

#解压
tar xf haproxy-2.4.3.tar.gz

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

#编辑文件
vi /etc/profile

#把这两项加上
export HAPROXY_HOME=/usr/app/haproxy-2.4.3
export PATH=$PATH:$HAPROXY_HOME/sbin

source /etc/profile

server rabbit1 rabbit1:5672 check inter 5000 rise 2 fall 3 weight 1

haproxy -f /etc/haproxy/haproxy.cfg

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

yum install -y keepalived

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  
    }
}

#!/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

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  
    }
}

systemctl start  keepalived

systemctl stop keepalived

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