RabbitMQ從零到集群高可用.NetCore(.NET5)-高可用集群構(gòu)建落地
2021-10-22 09:52:26

hostnamectl set-hostname rabbit1 --static

hostnamectl set-hostname rabbit2 --static

hostnamectl set-hostname rabbit3 --static

#修改hosts，因?yàn)閞abbitmq通訊要通過(guò)hostname
vi /etc/hosts

# 重啟網(wǎng)絡(luò)
systemctl restart network
# 重啟機(jī)器
init 6

#第一步 運(yùn)行package cloud提供的erlang安裝腳本
curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash

#第二步 安裝erlang
yum install erlang

#第三步 查看erlang版本號(hào),在命令行直接輸入erl
erl

#第一步 先導(dǎo)入兩個(gè)key
rpm --import https://packagecloud.io/rabbitmq/rabbitmq-server/gpgkey
rpm --import https://packagecloud.io/gpg.key

#第二步 運(yùn)行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依賴(lài)
yum -y install epel-release
yum -y install socat

#第六步 安裝
rpm -ivh rabbitmq-server-3.9.5-1.el8.noarch.rpm

#第七步 啟用管理平臺(tái)插件，啟用插件后，可以可視化管理RabbitMQ
rabbitmq-plugins enable rabbitmq_management

#第八步 啟動(dòng)應(yīng)用
systemctl start rabbitmq-server

#創(chuàng)建管理員賬戶
rabbitmqctl add_user admin 123456
#設(shè)置注冊(cè)的賬戶為管理員
rabbitmqctl set_user_tags admin administrator
#授權(quán)遠(yuǎn)程訪問(wèn)
rabbitmqctl set_permissions -p / admin "." "." ".*"
#重啟服務(wù)
systemctl restart rabbitmq-server

#查看防火墻狀態(tài)
systemctl status firewalld
#關(guān)閉防火墻
systemctl stop firewalld

#停止全部服務(wù)
systemctl stop rabbitmq-server

#給600權(quán)限
chmod 600 /var/lib/rabbitmq/.erlang.cookie

#編輯文件
vi /var/lib/rabbitmq/.erlang.cookie

#開(kāi)啟全部服務(wù)
systemctl start rabbitmq-server

# 1.停止服務(wù)
rabbitmqctl stop_app
# 2.重置狀態(tài)（需要更改節(jié)點(diǎn)類(lèi)型的時(shí)候執(zhí)行，首次不需要執(zhí)行，除非你節(jié)點(diǎn)是以disk加入集群的）
rabbitmqctl reset
# 3.節(jié)點(diǎn)加入
#rabbitmqctl join_cluster --ram rabbit@rabbit1
rabbitmqctl join_cluster rabbit@rabbit1
# 4.啟動(dòng)服務(wù)
rabbitmqctl start_app

/// <summary>
        /// 獲取集群連接對(duì)象
        /// </summary>
        /// <returns></returns>
        public static IConnection GetClusterConnection()
        {
            var factory = new ConnectionFactory
            {
                UserName = "admin",//賬戶
                Password = "123456",//密碼
                VirtualHost = "/" //虛擬機(jī)
            };
            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>
        /// 工作隊(duì)列模式
        /// </summary>
        public static void WorkerSendMsg()
        {
            string queueName = "worker_order";//隊(duì)列名
            //創(chuàng)建連接
            using (var connection = RabbitMQHelper.GetClusterConnection())
            {
                //創(chuàng)建信道
                using (var channel = connection.CreateModel())
                {
                    //創(chuàng)建隊(duì)列
                    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);
                        //發(fā)送消息到rabbitmq
                        channel.BasicPublish(exchange: "", routingKey: queueName, mandatory: false, basicProperties: properties, body);
                        Console.WriteLine($"發(fā)送消息到隊(duì)列:{queueName},內(nèi)容:{message}");
                    }
                }
            }
        }

#停止服務(wù)
rabbitmqctl stop_app

public static void WorkerConsumer()
        {
            string queueName = "worker_order";
            var connection = RabbitMQHelper.GetClusterConnection();
            {
                //創(chuàng)建信道
                var channel = connection.CreateModel();
                {
                    //創(chuàng)建隊(duì)列
                    channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    var consumer = new EventingBasicConsumer(channel);
                    ///prefetchCount:1來(lái)告知RabbitMQ,不要同時(shí)給一個(gè)消費(fèi)者推送多于 N 個(gè)消息，也確保了消費(fèi)速度和性能
                    ///global：是否設(shè)為全局的
                    ///prefetchSize:單條消息大小，通常設(shè)0，表示不做限制
                    //是autoAck=false才會(huì)有效
                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true);
                    int i = 1;
                    int index = new Random().Next(10);
                    consumer.Received += (model, ea) =>
                    {
                        //處理業(yè)務(wù)
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine($"{i},消費(fèi)者:{index},隊(duì)列{queueName}消費(fèi)消息長(zhǎng)度:{message.Length}");
                        Thread.Sleep(1000);
                        channel.BasicAck(ea.DeliveryTag, false); //消息ack確認(rèn)，告訴mq這條隊(duì)列處理完，可以從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

#進(jìn)入haproxy-2.4.3目錄后執(zhí)行
make TARGET=linux-glibc  PREFIX=/usr/app/haproxy-2.4.3
make install PREFIX=/usr/app/haproxy-2.4.3

#編輯文件
vi /etc/profile

#把這兩項(xiàng)加上
export HAPROXY_HOME=/usr/app/haproxy-2.4.3
export PATH=$PATH:$HAPROXY_HOME/sbin

source /etc/profile

global
    # 日志輸出配置、所有日志都記錄在本機(jī)，通過(guò) local0 進(jìn)行輸出
    log 127.0.0.1 local0 info
    # 最大連接數(shù)
    maxconn 4096
    # 改變當(dāng)前的工作目錄
    chroot /usr/app/haproxy-2.4.3
    # 以指定的 UID 運(yùn)行 haproxy 進(jìn)程
    uid 99
    # 以指定的 GID 運(yùn)行 haproxy 進(jìn)程
    gid 99
    # 以守護(hù)進(jìn)行的方式運(yùn)行
    daemon
    # 當(dāng)前進(jìn)程的 pid 文件存放位置
    pidfile /usr/app/haproxy-2.4.3/haproxy.pid

# 默認(rèn)配置
defaults
    # 應(yīng)用全局的日志配置
    log global
    # 使用4層代理模式，7層代理模式則為"http"
    mode tcp
    # 日志類(lèi)別
    option tcplog
    # 不記錄健康檢查的日志信息
    option dontlognull
    # 3次失敗則認(rèn)為服務(wù)不可用
    retries 3
    # 每個(gè)進(jìn)程可用的最大連接數(shù)
    maxconn 2000
    # 連接超時(shí)
    timeout connect 5s
    # 客戶端超時(shí)
    timeout client 120s
    # 服務(wù)端超時(shí)
    timeout server 120s

# 綁定配置
listen rabbitmq_cluster
    bind :5671
    # 配置TCP模式
    mode tcp
    # 采用加權(quán)輪詢的機(jī)制進(jìn)行負(fù)載均衡
    balance roundrobin
    # RabbitMQ 集群節(jié)點(diǎn)配置
    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

# 配置監(jiān)控頁(yè)面
listen monitor
    bind :8100
    mode http
    option httplog
    stats enable
    stats uri /stats
    stats refresh 5s

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",//賬號(hào)
                Password = "123456",//密碼
                VirtualHost = "/" //虛擬主機(jī)
            };

            return factory.CreateConnection();
        }

yum install -y keepalived

global_defs {
   # 路由id,主備節(jié)點(diǎn)不能相同
   router_id node1
}

# 自定義監(jiān)控腳本
vrrp_script chk_haproxy {
    # 腳本位置
    script "/etc/keepalived/haproxy_check.sh" 
    # 腳本執(zhí)行的時(shí)間間隔
    interval 5 
    weight 10
}

vrrp_instance VI_1 {
    # Keepalived的角色，MASTER 表示主節(jié)點(diǎn)，BACKUP 表示備份節(jié)點(diǎn)
    state MASTER  
    # 指定監(jiān)測(cè)的網(wǎng)卡，可以使用 ip addr 進(jìn)行查看
    interface ens33
    # 虛擬路由的id，主備節(jié)點(diǎn)需要設(shè)置為相同
    virtual_router_id 1
    # 優(yōu)先級(jí)，主節(jié)點(diǎn)的優(yōu)先級(jí)需要設(shè)置比備份節(jié)點(diǎn)高
    priority 100 
    # 設(shè)置主備之間的檢查時(shí)間，單位為秒 
    advert_int 1 
    # 定義驗(yàn)證類(lèi)型和密碼
    authentication { 
        auth_type PASS
        auth_pass 123456
    }

    # 調(diào)用上面自定義的監(jiān)控腳本
    track_script {
        chk_haproxy
    }

    virtual_ipaddress {
        # 虛擬IP地址，可以設(shè)置多個(gè)
        172.16.2.200  
    }
}

#!/bin/bash
# 判斷haproxy是否已經(jīng)啟動(dòng)
if [ `ps -C haproxy --no-header | wc -l` -eq 0 ] ; then
    #如果沒(méi)有啟動(dòng)，則啟動(dòng)
    haproxy -f /etc/haproxy/haproxy.cfg
fi

#睡眠3秒以便haproxy完全啟動(dòng)
sleep 3

#如果haproxy還是沒(méi)有啟動(dòng)，此時(shí)需要將本機(jī)的keepalived服務(wù)停掉，以便讓VIP自動(dòng)漂移到另外一臺(tái)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,主備節(jié)點(diǎn)不能相同    
   router_id node2  

}

vrrp_script chk_haproxy {
    script "/etc/keepalived/haproxy_check.sh" 
    interval 5 
    weight 10
}

vrrp_instance VI_1 {
    # BACKUP 表示備份節(jié)點(diǎn)
    state BACKUP 
    interface ens33
    virtual_router_id 1
    # 優(yōu)先級(jí)，備份節(jié)點(diǎn)要比主節(jié)點(diǎn)低
    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",//賬號(hào)
                Password = "123456",//密碼
                VirtualHost = "/" //虛擬主機(jī)
            };

            return factory.CreateConnection();
        }