[RabbitMQ] RabbitMQ运维问题

🌸个人主页:https://blog.csdn.net/2301_80050796?spm=1000.2115.3001.5343
🏵️热门专栏:
🧊 Java基本语法(97平均质量分)https://blog.csdn.net/2301_80050796/category_12615970.html?spm=1001.2014.3001.5482
🍕 Collection与数据结构 (93平均质量分)https://blog.csdn.net/2301_80050796/category_12621348.html?spm=1001.2014.3001.5482
🧀线程与网络(96平均质量分) https://blog.csdn.net/2301_80050796/category_12643370.html?spm=1001.2014.3001.5482
🍭MySql数据库(93平均质量分)https://blog.csdn.net/2301_80050796/category_12629890.html?spm=1001.2014.3001.5482
🍬算法(97平均质量分)https://blog.csdn.net/2301_80050796/category_12676091.html?spm=1001.2014.3001.5482
🍃 Spring(97平均质量分)https://blog.csdn.net/2301_80050796/category_12724152.html?spm=1001.2014.3001.5482
🎃Redis(97平均质量分)https://blog.csdn.net/2301_80050796/category_12777129.html?spm=1001.2014.3001.5482
🐰RabbitMQ(97平均质量分) https://blog.csdn.net/2301_80050796/category_12792900.html?spm=1001.2014.3001.5482
感谢点赞与关注~~~

一台RabbitMQ服务器接收消息的数量是有限的,假如一个RabbitMQ的服务器每秒可以接受消息的数量是1000条,但是假如我们每秒钟需要接收10万条消息的时候,这时候我们就必须通过搭建多个RabbitMQ结点来解决问题.
RabbitMQ集群允许消费者和生产者在RabbitMQ单个节点崩溃的时候继续运行,当失去一个RabbitMQ结点的时候,客户端可以能过重新连接到集群中的其他任何节点并继续生产或者消费.
接下来我们来介绍如何有效搭建一个RabbitMQ集群.

1. 多机多节点

需要注意的是我们在搭建一个具有多态服务器的集群的时候,需要这些服务器在同一个局域网中.我们使用三台云服务器来搭建集群.

首先我们需要在三台服务器上安装RabbitMQ程序.安装见前面的文章.
之后我们为了节点和节点之间相互识别,用来构成一个集群,我们需要在每个节点的hosts文件中**配置节点IP地址和节点的名称.
首先我们需要找到每个节点的hosts文件,并查看.

vim /etc/hosts

之后我们需要在hosts文件中配置其他节点的相关信息.格式为IP+主机名称.

#rabbitmq
10.0.0.232 iZ2vc7a1n9gvhfp589oav8Z
10.0.0.233 iZ2vc7a1n9gvhfp589oav6Z
10.0.0.234 iZ2vc7a1n9gvhfp589oav7Z

这里我们可以通过more /etc/hostname指令来查看每个主机的名称.

配置Erlang Cookie
RabbitMQ结点之间使用Cookie来进行身份验证,确认他们之间是否被允许相互通信,为了使得两个结点之间可以相互进行通信,我们==必须使得每个节点之间具有相同的共享密钥,称为Erlang Cookie.这是一个字符串,通常存储在本地文件中.
在RabbitMQ启动的时候,Erlang虚拟机会自动创建该文件,通常位于/var/lib/rabbitmq/.erlang.cookie和 $HOME/.erlang.cookie.
首先停止掉所有的结点服务.

systemctl stop rabbitmq-server

之后配置Erlang Cookie,只需要把结点上的.erlang.cookie文件分别拷贝到另外的两个结点上就可以.

#拷贝node3节点的文件到node1
scp /var/lib/rabbitmq/.erlang.cookie 
root@iZ2vc7a1n9gvhfp589oav8Z:/var/lib/rabbitmq/
#拷贝node3节点的文件到node2
scp /var/lib/rabbitmq/.erlang.cookie 
root@iZ2vc7a1n9gvhfp589oav6Z:/var/lib/rabbitmq/

拓展: Linux操作系统中的scp指令是用来在本地主机之间和远程主机之间进行安全文件拷贝的指令,基本语法格式是scp 本地文件路径远程用户名@远程主机名:远程文件路径.

配置好之后重新启动结点即可.

rabbit-server -datached

构建集群
为了将三个结点都连接起来,我们需要告诉另外两个结点加入拎一个结点,比如node1和node2结点加入node3结点.
但是在加入node3之前,我们必须对两个加入的集群进行重置操作,即删除结点上之前存在的所有资源和数据:

#1. 关闭RabbitMQ服务
rabbitmqctl stop_app
#2. 重置当前节点
rabbitmqctl reset
#3.加入节点 后⾯跟的是node3节点
rabbitmqctl join_cluster rabbit@iZ2vc7a1n9gvhfp589oav7Z
#4. 启动服务
rabbitmqctl start_app

常见问题
如果在管理界面中,发现集群中有的结点出现的Node statistics not available的情况,说明该结点在web上的管理插件还没有启用.

我们直接在节点上启动插件即可.

rabbitmq-plugins enable rabbitmq_management

2. 单机多结点

我们一般情况之下,我们只有一台服务器,我们可以使用单机多结点的方式来搭建集群,这样我们就可以不用多台服务器也可以验证集群中的某些特性.
和我们学习Redis中的集群一样,虚拟机只有一个,启动N个结点,结点和节点之间使用端口号来区分.

首先我们需要确定RabbitMQ运行没有问题.

root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl status
Status of node rabbit@iZ2ze9pwr3i8b65w9dr55dZ ...
Runtime

再启动两个结点.端口号分别是5673和5674.

RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management 
listener [{port,15673}]" RABBITMQ_NODENAME=rabbit2 rabbitmq-server -detached
RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management 
listener [{port,15674}]" RABBITMQ_NODENAME=rabbit3 rabbitmq-server -detached

接下来我们验证两个端口是否开通成功.注意放开云服务器中的端口防火墙.

在这里插入图片描述

3. 接下来我们就可以开始搭建集群了,首先我们停止RabbitMQ2和RabbitMQ3并重置.

root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit2 stop_app
Stopping rabbit application on node rabbit2@iZ2ze9pwr3i8b65w9dr55dZ ...
root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit2 reset
Resetting node rabbit2@iZ2ze9pwr3i8b65w9dr55dZ ...
root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit3 stop_app
Stopping rabbit application on node rabbit3@iZ2ze9pwr3i8b65w9dr55dZ ...
root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit3 reset
Resetting node rabbit3@iZ2ze9pwr3i8b65w9dr55dZ ...

把rabbit2和rabbit3添加到集群中,并重新启动.其中rabbit@iZ2ze9pwr3i8b65w9dr55dZ是主节点的主机名

root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit2 join_cluster rabbit@iZ2ze9pwr3i8b65w9dr55dZ
Clustering node rabbit2@iZ2ze9pwr3i8b65w9dr55dZ with rabbit@iZ2ze9pwr3i8b65w9dr55dZ
root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit3 join_cluster rabbit@iZ2ze9pwr3i8b65w9dr55dZ
Clustering node rabbit3@iZ2ze9pwr3i8b65w9dr55dZ with rabbit@iZ2ze9pwr3i8b65w9dr55dZ
root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit2 start_app
Starting node rabbit2@iZ2ze9pwr3i8b65w9dr55dZ ...
root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit3 start_app
Starting node rabbit3@iZ2ze9pwr3i8b65w9dr55dZ ...

查看集群状态

root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl cluster_status -n rabbit
Cluster status of node rabbit@iZ2ze9pwr3i8b65w9dr55dZ ...
Basics

Cluster name: rabbit@iZ2ze9pwr3i8b65w9dr55dZ

Disk Nodes

rabbit2@iZ2ze9pwr3i8b65w9dr55dZ
rabbit3@iZ2ze9pwr3i8b65w9dr55dZ
rabbit@iZ2ze9pwr3i8b65w9dr55dZ

Running Nodes

rabbit2@iZ2ze9pwr3i8b65w9dr55dZ
rabbit3@iZ2ze9pwr3i8b65w9dr55dZ
rabbit@iZ2ze9pwr3i8b65w9dr55dZ

通过集群中的主节点可以查看到其他结点的状态.
在这里插入图片描述

3. 结点宕机演示

集群搭建好之后,也会存在一定的问题,结点在宕机的时候,其他结点的数据不会在进行同步,我们下面来进行演示:

首先添加队列,rabbit结点添加一个,rabbit2添加一个.
添加之后,集群中的其他结点中也会把队列的信息进行同步.

注意这里与我们前面的Redis的主从模式不同,RabbitMQ的主从节点都可以写入信息,也会在其他的结点中进行同步,但是Redis只能在主节点中进行写数据,从结点中只可以读数据.

往testQueue队列中发送一条消息.发送之后其他从节点的队列中也会有消息.
接下来我们进行宕机演示,我们关闭rabbit(主)结点.

两个从结点的消息队列和队列中的信息均已经丢失.
如何解决这种消息丢失的问题呢,我们就需要引入仲裁队列.

4. 仲裁队列

RabbitMQ的仲裁队列是一种基于Raft一致性算法实现持久化,复制的FIFO队列,使用仲裁队列可以在RabbitMQ结点之间进行数据的复制,从而达到一个结点宕机的时候,队列仍然可以提供服务的效果.Raft算法我们在之前Redis的文章中曾经介绍过.

4.1 Raft协议

Raft是⼀种用于管理和维护分布式系统一致性的协议,它是一种共识算法,旨在实现高可用性和数据的持久性.Raft通过在节点间复制数据来保证分布式系统中的一性，即使在节点故障的情况下也能保证数据不会丢失.
共识算法（Consensus Algorithm）,它允许多个分布式节点就某个值或一系列值达成⼀致性协议.即使在⼀些节点发生故障,网络分区或其他问题的情况下,共识算法也能保证系统的一致性和数据的可靠性.接下来我们就来详细介绍一下Raft算法.
Raft使用Quorum机制来实现共识和容错,我们将对Raft集群的操作必须得到大多数(大于总结点数的一半)结点的同意才可以提交.在RabbitMQ集群中,该算法一般用来选取主节点.

结点角色
Raft集群必须存在一个主节点(leader),客户端向集群发起的所有操作都必须经过主节点的处理,所以Raft的核心算法中的最重要的部分就是选主.没有主节点集群就无法工作.在执行了选主过程之后,集群中每一个结点都会识别出一个特定的,唯一的leader.

在Raft算法中,每个节点都处于一下的三种角色之一.

leader(领导) : 负责处理所有客户的情况,并将这些请求作为日志复制到Follower.Leader会定期向Follower发送心跳包,一维持领导者的地位,防止Follower进入选举过程.
follower(跟随者): 接收来自Leader的日志条目,并在本地应用这些条目.跟随者不会直接处理客户端请求.
Candidate(候选者): 当在一段时间没有收到来自Leader的心跳包之后,他会变得不确定Leader是否任然可用,在这种情况之下,跟随者会转变为候选者,并开始尝试通过投票的方式成为新的Leader.
它们之间的转换关系如下:

任期
Raft将时间划分成任意长度的任期,每一段任期从一次选举开始,在这个时候会有一个或者是多个候选者会尝试成为Leader,在成功完成一次Leader的选举之后,该结点就会一直是集群的Leader直到任期结束,之后进行下一次的Leader选举.在某些情况之下一次选举无法选举出Leader,这个时候任期会以没有Leader而结束.同时进行新的任期(包含一次新的选举)会重新开始.

每个节点中都保存着一个current term,在通信的时候带上这个term的值.
每个结点中都存储着一个当前的任期号.该任期号回随着时间单调递增,结点之间通信的时候会交换当前的任期号,**如果一个结点当前的任期号比其他结点小,那么他就将自己的任期号更新为较大的哪个值,如果一个Candidate或者Leader发现者自己的任期号已经过期,会立即回到follower状态.如果一个结点接收了一个带着过期的任期号的请求,那么他就会拒绝这次的请求.

Raft算法中服务器节点之间主要采用RPC进行通信,主要有两类RPC请求:

RequestVote RPCs: 请求投票,由candidate在选举过程中发出.
AppendEntries RPCs:追加条目,由leader发出,用来做日志复制和提供心跳机制,即心跳包.

选举详细过程
Raft采用心跳机制来触发Leader选举,当服务器启动的时候,都是follower状态,如果follower在指定的时间之内没有收到来自Leader的心跳(有三种情况,一是还没有选举出Leader,二是Leader宕机,三是Leader与follower之间网络发生故障),则会主动发起选举.

率先超时的结点,自增当前的任期号之后切换为候选者状态,并给自己投一票.以并行的方式向其他结点发送一个投票请求.之后等待其他结点的回复.

这个过程中,各个结点可能出现三种结果
- 赢得选举,成为Leader.
- 其他结点赢得选举,他自行切换到follower.
- 一段时间之内没有收到投票,保持候选者状态,继续发出选举.

投票要求

每个服务器节点会按照先来后到的原则只投给一个候选者.每个节点手中只有一票的权利.

接下来我们对这三种情况进行说明:

第一种情况,赢得选举之后,新的Leader会立刻给所有的结点发送消息,广而告之,避免其他的结点触发新的选举.
第二种情况,比如前面有三个结点ABC同时发起选举,而A的选举消息先到达了C,C给A投出了一票,当B到达C的时候,C手中已经没有投票的权利了,这时候A就胜出了,A胜出之后,会给BC发送心跳消息,结点B发现A的term不低于自己的term,知道已经有了Leader了,于是把自己转换成为了follower.
第三种情况: 没有任何节点可以获得最多的投票,比如所有的follower同时变为了Candidate,然后他们的票都会投给自己,这样就都没有Candidate能得到的超过半数的投票了.当这种情况发生的时候,每个Candidate都会进行一次超时响应,然后通过自增任期号来开启新一轮选举,如果没有额外的措施,这种结果会一直持续下去.

为了可以解决上述的问题,Raft采用随机选取超时时间来确保很少产生无结果的投票.为了防止投票一开始就会被瓜分掉,选举时间是从一个固定的区间中随机选择.这样可以把结点之间分散开来一确保在大多数的情况下会只有一个服务器率先超时,那么这个时候,他就可以赢得选举并在其他服务器结束超时之前发送心跳.

4.2 Raft协议之下的消息复制

每个仲裁队列都有多个副本,他包含一个主和多个从副本.replication factor为5的仲裁队列将会有一个主副本和4个从副本.每个副本都在不同的RabbitMQ结点上.
生产者和消费者只会与主副本进行交互,主副本在将这些命令复制到从副本,当主副本所在的结点宕机的时候,其中一个从副本会被选举成为主副本,继续提供服务.
在这里插入图片描述

5. 仲裁队列的使用

创建仲裁队列

使用Spring框架进行创建

@Bean
public Queue quorumQueue(){
    return QueueBuilder.durable("quorum_queue").quorum().build();
}

在管理界面进行创建

接收发送消息
仲裁队列接收发送消息和普通的队列是一样的.
宕机演示

首先创建一个仲裁队列
给仲裁队列发送一条消息

三个结点的仲裁队列都存储了消息.

停止主副本所在的结点

root@iZ2ze9pwr3i8b65w9dr55dZ:~# rabbitmqctl -n rabbit stop_app
Stopping rabbit application on node rabbit@iZ2ze9pwr3i8b65w9dr55dZ ...

停掉之后我们发现其他结点的仲裁队列仍然存在.

6. HAProxy负载均衡

当我们使用集群的策略来对负载能力做进一步的提升的时候,但这里还存在一些问题.
想一想我们在写代码的时候,集群中有三个结点,我们在写代码的时候,该访问那个结点呢?当然是那个结点都可以.这时候就存在两个问题:

如果我们访问的是node1,node1挂了,程序会出现问题,所以最好是有一个统一的入口,一个结点故障的时候,流量可以及时转移到其他的结点.
如果所有的客户端都与node1连接,那么node1的网络负载必然会大大增加,而其他结点由于没有那么多的负载而造成了资源的浪费.
这时候负载均衡显得格外的重要.

引入负载均衡之后,我们在写代码的过程中,就会统一连接负载均衡所在的端口号,由负载均衡来决定一个客户端该和集群中的那一个节点来连接,当有一个结点宕机的时候,负载均衡还会自动把消息重新分配到其他的结点上.
这里我们主要讲解的是使用HAProxy来实现负载均衡.

6.1 安装

首先我们需要在Linux操作系统中安装AHPorxy.

安装HAProxy

#更新软件包
sudo apt-get update
#安装HAProxy
sudo apt-get install haproxy

验证安装

root@iZ2ze9pwr3i8b65w9dr55dZ:~/lottery# sudo systemctl status haproxy
● haproxy.service - HAProxy Load Balancer
     Loaded: loaded (/lib/systemd/system/haproxy.service; enabled; vendor preset: enabled)
     Active: active (running) since Sun 2025-01-12 20:48:35 CST; 16min ago
       Docs: man:haproxy(1)
             file:/usr/share/doc/haproxy/configuration.txt.gz
    Process: 830 ExecStartPre=/usr/sbin/haproxy -Ws -f $CONFIG -c -q $EXTRAOPTS (code=exited, status=0/S>
   Main PID: 941 (haproxy)
      Tasks: 3 (limit: 1917)
     Memory: 71.9M
        CPU: 126ms
     CGroup: /system.slice/haproxy.service
             ├─ 941 /usr/sbin/haproxy -Ws -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -S /run/haprox>
             └─1065 /usr/sbin/haproxy -Ws -f /etc/haproxy/haproxy.cfg -p /run/haproxy.pid -S /run/haprox>

Jan 12 20:48:34 iZ2ze9pwr3i8b65w9dr55dZ systemd[1]: Starting HAProxy Load Balancer...
Jan 12 20:48:35 iZ2ze9pwr3i8b65w9dr55dZ haproxy[941]: [NOTICE]   (941) : New worker #1 (1065) forked
Jan 12 20:48:35 iZ2ze9pwr3i8b65w9dr55dZ systemd[1]: Started HAProxy Load Balancer.

修改haporxy.cfg
vim /etc/haproxy/haproxy.cfg
追加一下内容:

# haproxy web 管理界⾯
listen stats
	bind *:8100 
	mode http
	stats enable
	stats realm Haproxy\ Statistics 
	stats uri /
	stats auth admin:admin
# 配置负载均衡
listen rabbitmq
	bind *:5670
	mode tcp
	balance roundrobin
	server rabbitmq1 127.0.0.1:5672 check inter 5000 rise 2 fall 3
	server rabbitmq2 127.0.0.1:5673 check inter 5000 rise 2 fall 3
	server rabbitmq3 127.0.0.1:5674 check inter 5000 rise 2 fall 3

在HAProxy的配置文件中,listen stats是用来设置一个监听器的指令,这个监听器就是专门用户HAProxy的统计信息页面.这些统计信息可以通过web界面来访问.
4. 重启HAProxy
5. 查看HAProxy
在这里插入图片描述

6.2 使用

修改配置文件
引入HAProxy之后,RabbitMQ的集群使用和单机使用方式⼀样,只不过需要把RabbitMQ的IP和port改为HAProxy的IP和port.

host: 182.92.204.253
port: 5670

声明队列

@Bean
public Queue clusterQueue(){
    return QueueBuilder.durable("cluster_queue").quorum().build();
}

发送消息

@RequestMapping("/cluster")
public String cluster(){
    rabbitTemplate.convertAndSend("","cluster_queue","test...");
    return "发送成功";
}