亲宝软件园·资讯

展开

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

包子wxl 人气:0
想了解运用.net core中实例讲解RabbitMQ高可用集群构建的相关内容吗,包子wxl在本文为您仔细讲解.net core RabbitMQ集群构建的相关知识和一些Code实例,欢迎阅读和指正,我们先划重点:.net,core,RabbitMQ集群构建,下面大家一起来学习吧。

一、集群架构简介

当单台 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

二、普通集群搭建

2.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)安装前,先修改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

2)把ip对应hostname添加到后面

172.16.2.84 rabbit1
172.16.2.85 rabbit2
172.16.2.86 rabbit3

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

3)安装Erlang

#第一步 运行package cloud提供的erlang安装脚本
curl -s http://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash

#第二步 安装erlang
yum install erlang

#第三步 查看erlang版本号,在命令行直接输入erl
erl

4)安装RabbitMQ

#第一步 先导入两个key
rpm --import http://packagecloud.io/rabbitmq/rabbitmq-server/gpgkey
rpm --import http://packagecloud.io/gpg.key

#第二步 运行package cloud提供的rabbitmq安装脚本
curl -s http://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/

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)拷贝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

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管理界面(随便打开一个都是一样的)。

2.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环境搭建。

1)下载

HAProxy 官方下载地址为:https://www.haproxy.org/,如果这个网站无法访问,也可以从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

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

3)配置环境变量

#编辑文件
vi /etc/profile

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

使得配置的环境变量立即生效:

source /etc/profile

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 用于指定节点在轮询过程中的权重。

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

1)安装

yum install -y keepalived

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相同

3)启动服务

分别在KeepAlived1和KeepAlived2上启动KeepAlived服务,命令如下:

systemctl start  keepalived

启动后此时 keepAlived1 为主节点,可以在keepAlived1 上使用ip a命令查看到虚拟 IP 的情况:

此时只有 keepAlived1上是存在虚拟 IP 的,而keepAlived2 上是没有的。

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

加载全部内容

相关教程
猜你喜欢
用户评论