RabbitMQ 集群篇_mq集群

RabbitMQ 集群篇

00、集群架构原理

前面我们有介绍到 RabbitMQ 内部有各种基础构件，包括队列、交换器、绑定、虚拟主机等，他们组成了 AMQP 协议消息通信的基础，而这些构件以元数据的形式存在，它始终记录在 RabbitMQ 内部，它们分别是：

• 队列元数据：队列名称和它们的属性
• 交换器元数据：交换器名称、类型和属性
• 绑定元数据：一张简单的表格展示了如何将消息路由到队列
• vhost 元数据：为 vhost 内的队列、交换器和绑定提供命名空间和安全属性

PS：元数据，指的是包括队列名字属性、交换机的类型名字属性、绑定信息、vhost等基础信息，不包括队列中的消息数据。

这些元数据，其实本质是一张查询表，里面包括了交换器名称和一个队列的绑定列表，当你将消息发布到交换器中，实际上是将你所在的信道将消息上的路由键与交换器的绑定列表进行匹配，然后将消息路由出去。

有了这个机制，那么在所有节点上传递交换器消息将简单很多，而 RabbitMQ 所做的事情就是把交换器元数据拷贝到所有节点上，因此每个节点上的每条信道都可以访问完整的交换器。

如果消息生产者所连接的是节点 2 或者节点 3，此时队列1的完整数据不在该两个节点上，那么在发送消息过程中这两个节点主要起了一个路由转发作用，根据这两个节点上的元数据转发至节点1上，最终发送的消息还是会存储至节点1的队列1上。

同样，如果消息消费者所连接的节点2或者节点3，那这两个节点也会作为路由节点起到转发作用，将会从节点1的队列1中拉取消息进行消费。

与常见的集群主从架构模式不同在于：RabbitMQ集群模式下，仅仅只是同步元数据，每个队列内容还是在自己的服务器节点上

这么设计主要还是基于集群本身的性能和存储空间上来考虑：

• 存储空间：真正存放数据的地方是在队列里面，如果每个集群节点都拥有所有队列的完全数据拷贝，那么每个节点的存储空间会非常大，集群的消息积压能力会非常弱。例如你现在存储了 3G 队列内容，那么在另外一个只有 1G 存储空间的节点上，就会造成内存空间不足的情况，也就是无法通过集群节点的扩容提高消息积压能力。
• 性能：消息的发布者需要将消息复制到每一个集群节点，每一条消息都会触发磁盘活动，这会导致整个集群内性能负载急剧拉升。

既然每个队列内容还是在自己的服务器节点上，同样也会带来新的问题，那就是如果队列所在服务器挂了，那存在服务器上的队列数据是不是全部都丢失了？

在单个节点上，RabbitMQ 存储数据有两种方案：

• 内存模式：这种模式会将数据存储在内存当中，如果服务器突然宕机重启之后，那么附加在该节点上的队列和其关联的绑定都会丢失，并且消费者可以重新连接集群并重新创建队列；
• 磁盘模式：这种模式会将数据存储磁盘当中，如果服务器突然宕机重启，数据会自动恢复，该队列又可以进行传输数据了，并且在恢复故障磁盘节点之前，不能在其它节点上让消费者重新连到集群并重新创建队列，如果消费者继续在其它节点上声明该队列，会得到一个 404 NOT_FOUND 错误，这样确保了当故障节点恢复后加入集群，该节点上的队列消息不会丢失，也避免了队列会在一个节点以上出现冗余的问题。

在集群中的每个节点，要么是内存节点，要么是磁盘节点，如果是内存节点，会将所有的元数据信息仅存储到内存中，而磁盘节点则不仅会将所有元数据存储到内存上， 还会将其持久化到磁盘。

在单节点 RabbitMQ 上，仅允许该节点是磁盘节点，这样确保了节点发生故障或重启节点之后，所有关于系统的配置与元数据信息都会从磁盘上恢复。

而在 RabbitMQ 集群上，至少有一个磁盘节点，也就是在集群环境中需要添加 2 台及以上的磁盘节点，这样其中一台发生故障了，集群仍然可以保持运行。其它节点均设置为内存节点，这样会让队列和交换器声明之类的操作会更加快速，元数据同步也会更加高效。

01、集群的几种模式

参考文章：
https://www.cnblogs.com/rouqinglangzi/p/10584748.html

RabbitMQ模式大概分为以下四种：

• 普通集群模式/主备集群模式（Warren）：默认的模式

• 镜像队列模式（Mirror）：实际工作用的最多

• **远程模式（Shovel）：**属于分布式部署，这个属于早期常用的，现在很少用了

• **多活模式（Federation）：**属于分布式部署，一般异地集群部署会用这种模式

**RabbitMQ集群对延迟非常敏感，应该只在本地局域网中使用。**在广域网中不应该使用集群模式，而应该使用Federation或者Shovel来进行分布式部署。

01-1、主备模式（Warren）

也称为 Warren (兔子窝) 模式。实现 rabbitMQ 的高可用集群，一般在并发和数据量不高的情况下，这种模式非常的好用且简单。

默认情况下，RabbitMQ集群中队列的内容仅位于单个节点上（即声明该队列的节点，也就是主节点）。创建队列时，只会在某一个节点上创建队列，其它节点上并不含有队列，而只是含有创建节点的元数据(包括队列信息，绑定关系等)。如果队列所在的节点故障，则队列就崩溃了。

1、基本特征

• 交换机和队列的元数据存在于所有的节点上
• queue队列中的完整数据只存在于创建该队列的节点上
• 其他节点只保存队列的元数据信息以及指向当前队列的owner node的指针

2、数据消费

进行数据消费时随机连接到一个节点，当队列不是当前节点创建的时候，需要有一个从创建队列的实例拉取队列数据的开销。此外由于需要固定从单实例获取数据，因此会出现单实例的瓶颈。

3、优点：可以由多个节点消费单个队列的数据，提高了吞吐量

4、缺点：

• 节点实例需要拉取数据，因此集群内部存在大量的数据传输
• 可用性保障低，一旦创建队列的节点宕机，只有等到该节点恢复其他节点才能继续消费消息

02-2、镜像模式（Mirror）

集群模式非常经典的就是Mirror镜像模式，保证100%数据不丢失，在实际工作中也是用的最多的，而且实现集群也非常简单，一般互联网大厂都会构建这种镜像集群模式，原理主要是在主备的基础上进行了扩展，集群中所有的节点设备都是同步的，每一个队列，交换机里面的配置信息和我们的数据都是同步的，对于这些镜像在底层同时进行工作，前面的话采用一个负载均衡器，采用nginx或者haproxy也好，进行负载均衡。

1、基本特征

• 创建的queue，不论是元数据还是完整数据都会在每一个节点上保存一份
• 向queue中写消息时，都会自动同步到每一个节点上

2、优点：

• 保障了集群的高可用
• 配置方便，只需要在后台配置相应的策略，就可以将指定数据同步到指定的节点或者全部节点

3、缺点：

• 性能开销较大，网络带宽压力和消耗很严重，所以镜像队列的吞吐量会低于普通模式
• 无法线性扩展，例如单个queue的数据量很大，每台机器都要存储同样大量的数据

03-3、远程模式（Shovel）

远程模式可以实现双活的一种模式，简称 shovel 模式，所谓的 shovel 就是把消息进行不同数据中心的复制工作，可以跨地域的让两个MQ 集群互联，远距离通信和复制。这种模式配置比较复杂，属于RabbitMQ比较早期的模型了，现在很少使用了。

官方文档：Shovel。

例如说一个集群，我们都会放在一个机房里面，那么如果北京的机房出现了一些事故停电，或者自然灾害，那么这个集群都会宕机了，那么在我们对数据要求极高的大型应用我们需要设置多活或者双活的模式，也就是要搭建多个数据中心，或者多套集群，那么这些集群可以一个会放在上海，一个放在北京，还有应放在广州，三个集群数据都是同步的，中间有任何一个集群出现了问题，马上灵活的切换，那么这三个集群都是可以访问的话，我们可能会按照距离，或者访问速度来进行优先选择哪组集群，或者数据中心进行访问，所有多活模式，在银行开发的时候一般也叫做容灾的机制，至少构建两套集群放在不同的地域，一个有问题了，立马进行切换，不至于整个系统宕机，这就是多活模式，在多活模式中MQ也提供了相应的实现方式，早期使用的Shovel模式，这个模式是mq自带的一种模式，主要就是可以把消息在不同的数据中心进行负载分发，主要就是可以实现跨地域的让两个mq集群进行互联。

那么这个shovel模式需要统一的版本，网络达到一个什么样的水平，配置的话也是有些复杂，这种的话目前已经淘汰了，在真正的数据复制的情况下，会使用多活模式。

04-4、多活模式（Federation）

这种模式也是实现异地数据复制的主流模式，这种模式需要依赖 rabbitMQ的 Federation 插件，本身不是mq自带的东西，可以实现持续的、可靠的 AMQP 数据通信。因此配置起来也比较容易，相当于配置简单化之后的shovel。

官方文档：Federation

这是实现异地数据复制的主流模式，现在实现异地集群的一般都是采用这种：双活 或者 多活模型。

02、单机多实例搭建

**场景：**假设有三个rabbitmq节点，分别为rabbit-1、rabbit-2、rabbit-3，rabbit-1作为主节点，2和3作为从节点。
启动命令：RABBITMQ_NODE_PORT={{端口号}} RABBITMQ_NODENAME={{节点名}} rabbitmq-server -detached
结束命令：rabbitmqctl -n {{节点名}} stop

02-1、关闭RabbitMQ服务

RabbitMQ 默认是使用服务的启动的，单机多节点时需要改为手动启动，先停止运行中的RabbitMQ 服务

sudo systemctl stop rabbitmq-server
1

02-2、启动三个RabbitMQ节点

sudo RABBITMQ_NODE_PORT=5672 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15672}]" RABBITMQ_NODENAME=rabbit-1 rabbitmq-server -detached

sudo RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server -detached

sudo RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" RABBITMQ_NODENAME=rabbit-3 rabbitmq-server -detached
1
2
3
4
5

02-3、查看服务其启动状况

命令：ps aux|grep rabbitmq

[root@CentOS7 ~]# ps aux|grep rabbitmq
rabbitmq 13849 30.5  2.4 2296044 90600 ?       Sl   13:41   0:08 /usr/lib64/erlang/erts-11.1.8/bin/beam.smp -W w -MBas ageffcbf -MHas ageffcbf -MBlmbcs 512 -MHlmbcs 512 -MMmcs 30 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -sbwt none -sbwtdcpu none -sbwtdio none -B i -- -root /usr/lib64/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa  -noshell -noinput -s rabbit boot -boot start_sasl -rabbitmq_management listener [{port,15672}] -lager crash_log false -lager handlers [] -noshell -noinput
rabbitmq 13853  0.0  0.0   4356   544 ?        Ss   13:41   0:00 erl_child_setup 65535
rabbitmq 13876  0.0  0.0  48912   520 ?        S    13:41   0:00 /usr/lib64/erlang/erts-11.1.8/bin/epmd -daemon
rabbitmq 13897  0.0  0.0  11592   456 ?        Ss   13:41   0:00 inet_gethost 4
rabbitmq 13898  0.0  0.0  41552  1196 ?        S    13:41   0:00 inet_gethost 4
rabbitmq 13925 50.4  2.4 2298236 90540 ?       Sl   13:41   0:08 /usr/lib64/erlang/erts-11.1.8/bin/beam.smp -W w -MBas ageffcbf -MHas ageffcbf -MBlmbcs 512 -MHlmbcs 512 -MMmcs 30 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -sbwt none -sbwtdcpu none -sbwtdio none -B i -- -root /usr/lib64/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa  -noshell -noinput -s rabbit boot -boot start_sasl -rabbitmq_management listener [{port,15673}] -lager crash_log false -lager handlers [] -noshell -noinput
rabbitmq 13929  0.0  0.0   4356   532 ?        Ss   13:41   0:00 erl_child_setup 65535
rabbitmq 13973  0.0  0.0  11592   456 ?        Ss   13:41   0:00 inet_gethost 4
rabbitmq 13974  0.0  0.0  41552  1200 ?        S    13:41   0:00 inet_gethost 4
rabbitmq 14003 96.4  2.6 2306668 98156 ?       Sl   13:41   0:08 /usr/lib64/erlang/erts-11.1.8/bin/beam.smp -W w -MBas ageffcbf -MHas ageffcbf -MBlmbcs 512 -MHlmbcs 512 -MMmcs 30 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -sbwt none -sbwtdcpu none -sbwtdio none -B i -- -root /usr/lib64/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa  -noshell -noinput -s rabbit boot -boot start_sasl -rabbitmq_management listener [{port,15674}] -lager crash_log false -lager handlers [] -noshell -noinput
rabbitmq 14007  0.0  0.0   4356   528 ?        Ss   13:41   0:00 erl_child_setup 65535
rabbitmq 14051  0.0  0.0  11592   456 ?        Ss   13:41   0:00 inet_gethost 4
rabbitmq 14052  0.0  0.0  41552  1196 ?        S    13:41   0:00 inet_gethost 4

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

02-4、加入集群

1、重置rabbit-1为主节点：

# 停止应用
> sudo rabbitmqctl -n rabbit-1 stop_app
# 目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
> sudo rabbitmqctl -n rabbit-1 reset
# 启动应用
> sudo rabbitmqctl -n rabbit-1 start_app
1
2
3
4
5
6

2、将rabbit-2、rabbit-3加入到集群

# 停止应用
sudo rabbitmqctl -n rabbit-2 stop_app
# 目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
sudo rabbitmqctl -n rabbit-2 reset
# 将rabbit2节点加入到rabbit1（主节点）集群当中【节点名@主机名】--ram 表示这是一个内存节点
sudo rabbitmqctl -n rabbit-2 join_cluster rabbit-1@`hostname -s` --ram
sudo rabbitmqctl -n rabbit-2 start_app

# rabbit-3 节点重置后加入集群
sudo rabbitmqctl -n rabbit-3 stop_app
sudo rabbitmqctl -n rabbit-3 reset
# --disc表示磁盘节点（默认不写也是磁盘节点）
sudo rabbitmqctl -n rabbit-3 join_cluster rabbit-1@`hostname -s` --disc
sudo rabbitmqctl -n rabbit-3 start_app
1
2
3
4
5
6
7
8
9
10
11
12
13
14

02-5、验证集群状态

命令：sudo rabbitmqctl cluster_status -n rabbit-1

[root@CentOS7 ~]# rabbitmqctl cluster_status -n rabbit-1
Cluster status of node rabbit-1@CentOS7 ...
Basics

Cluster name: rabbit-1@CentOS7

Disk Nodes

rabbit-1@CentOS7
rabbit-3@CentOS7

RAM Nodes

rabbit-2@CentOS7

Running Nodes

rabbit-1@CentOS7
rabbit-2@CentOS7
rabbit-3@CentOS7

Versions

rabbit-1@CentOS7: RabbitMQ 3.8.14 on Erlang 23.2.7
rabbit-2@CentOS7: RabbitMQ 3.8.14 on Erlang 23.2.7
rabbit-3@CentOS7: RabbitMQ 3.8.14 on Erlang 23.2.7

Maintenance status

Node: rabbit-1@CentOS7, status: not under maintenance
Node: rabbit-2@CentOS7, status: not under maintenance
Node: rabbit-3@CentOS7, status: not under maintenance

Alarms

(none)

Network Partitions

(none)

Listeners

Node: rabbit-1@CentOS7, interface: [::], port: 15672, protocol: http, purpose: HTTP API
Node: rabbit-1@CentOS7, interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
Node: rabbit-1@CentOS7, interface: [::], port: 5672, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0
Node: rabbit-2@CentOS7, interface: [::], port: 15673, protocol: http, purpose: HTTP API
Node: rabbit-2@CentOS7, interface: [::], port: 25673, protocol: clustering, purpose: inter-node and CLI tool communication
Node: rabbit-2@CentOS7, interface: [::], port: 5673, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0
Node: rabbit-3@CentOS7, interface: [::], port: 15674, protocol: http, purpose: HTTP API
Node: rabbit-3@CentOS7, interface: [::], port: 25674, protocol: clustering, purpose: inter-node and CLI tool communication
Node: rabbit-3@CentOS7, interface: [::], port: 5674, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0

Feature flags

Flag: drop_unroutable_metric, state: enabled
Flag: empty_basic_get_metric, state: enabled
Flag: implicit_default_bindings, state: enabled
Flag: maintenance_mode_status, state: enabled
Flag: quorum_queue, state: enabled
Flag: user_limits, state: enabled
Flag: virtual_host_metadata, state: enabled

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

web页面查看：

02-6、解除集群

1、在自己节点重置当前节点，退出集群

# 停止服务、重置集群状态、重启服务
rabbitmqctl -n rabbit-2 stop_app && rabbitmqctl -n rabbit-2 reset && rabbitmqctl -n rabbit-2 start_app
1
2

2、在集群其他任意一个正常节点将节点踢出集群

# 必须先关闭当前节点才能把该节点解除集群
sudo rabbitmqctl -n rabbit-3@CentOS7 stop_app
# 在节点1上操作，把节点3踢出集群（可以在任意节点操作，不一定要在主节点）
sudo rabbitmqctl -n rabbit-1 forget_cluster_node rabbit-3@CentOS7
1
2
3
4

02-7、删除集群模式

# 1.暂停rabbitmq节点服务
sudo rabbitmqctl -n rabbit-3 stop
sudo rabbitmqctl -n rabbit-2 stop
sudo rabbitmqctl -n rabbit-1 stop
# 2.清空目录
rm -rf /var/lib/rabbitmq/mnesia/
1
2
3
4
5
6

访问web端需要新增账户：

rabbitmqctl -n rabbit-1 add_user admin admin && rabbitmqctl -n rabbit-1 set_user_tags admin administrator && rabbitmqctl -n rabbit-1 set_permissions -p / admin ".*" ".*" ".*"

rabbitmqctl -n rabbit-2 add_user admin admin && rabbitmqctl -n rabbit-2 set_user_tags admin administrator &&rabbitmqctl -n rabbit-2 set_permissions -p / admin ".*" ".*" ".*"

rabbitmqctl -n rabbit-3 add_user admin admin && rabbitmqctl -n rabbit-3 set_user_tags admin administrator &&rabbitmqctl -n rabbit-3 set_permissions -p / admin ".*" ".*" ".*"
1
2
3
4
5

03、多机多节点部署

Tips：
如果采用多机部署方式，需读取其中一个节点的cookie, 并复制到其他节点（节点之间通过cookie确定相互是否可通信）。cookie存放在/var/lib/rabbitmq/.erlang.cookie。
例如：主机名分别为rabbit-1、rabbit-2
1、逐个启动各节点
2、配置各节点的hosts文件( vim /etc/hosts)
ip1：rabbit-1
ip2：rabbit-2
其它步骤雷同单机部署方式

03-1、环境准备

准备三台服务器CentOS 7.8：192.168.3.12、192.168.3.13、192.168.3.14

提示：如果使用虚拟机，可以在一台VM上安装好RabbitMQ后，创建快照，从快照创建链接克隆，会节省很多磁盘空间

03-2、初始化环境

1、分别修改三台主机名：

hostnamectl set-hostname rabbit-1
hostnamectl set-hostname rabbit-2
hostnamectl set-hostname rabbit-3
1
2
3

2、修改每台机器的 /etc/hosts 文件

cat >> /etc/hosts <<EOF
192.168.3.12 rabbit-1
192.168.3.13 rabbit-2
192.168.3.14 rabbit-3
EOF
1
2
3
4
5

3、同步三台直接的.erlang.cookie信息（拷贝.cookie时，各节点都必须停止MQ服务）

scp /var/lib/rabbitmq/.erlang.cookie root@rabbit-2:/var/lib/rabbitmq/
scp /var/lib/rabbitmq/.erlang.cookie root@rabbit-3:/var/lib/rabbitmq/
1
2

由于你可能在三台主机上使用不同的账户进行操作，为避免后面出现权限不足的问题，这里建议将 cookie 文件原来的 400 权限改为 600，命令如下：

chmod 600 /var/lib/rabbitmq/.erlang.cookie
1

03-3、启动服务加入集群

1、首先每台机器开启防火墙

# 添加端口
sudo firewall-cmd --zone=public --add-port=4369/tcp --permanent
sudo firewall-cmd --zone=public --add-port=5672/tcp --permanent
sudo firewall-cmd --zone=public --add-port=25672/tcp --permanent
sudo firewall-cmd --zone=public --add-port=15672/tcp --permanent

# 重启防火墙
sudo firewall-cmd --reload
1
2
3
4
5
6
7
8

2、启动每台机器RabbitMQ服务

sudo systemctl start rabbitmq-server
或者
rabbitmq-server -detached
1
2
3

3、重置节点1

# 停止RabbitMQ 应用
rabbitmqctl stop_app
# 重置RabbitMQ 设置
rabbitmqctl reset
# 启动RabbitMQ 应用
rabbitmqctl start_app
1
2
3
4
5
6

4、重置节点2 并加入集群：

# 停止RabbitMQ 应用
rabbitmqctl stop_app
# 重置RabbitMQ 设置
rabbitmqctl reset
# 加入到集群
rabbitmqctl join_cluster rabbit@rabbit-1 --ram
# 启动RabbitMQ 应用
rabbitmqctl start_app
1
2
3
4
5
6
7
8

5、重置节点3 并加入集群：

# 停止 RabbitMQ 应用
rabbitmqctl stop_app
# 重置 RabbitMQ 设置
rabbitmqctl reset
# 节点加入到集群
rabbitmqctl join_cluster rabbit@rabbit-3 --ram
# 启动 RabbitMQ 应用
rabbitmqctl start_app
1
2
3
4
5
6
7
8

6、查看集群（也可以直接通过web页面查看）

rabbitmqctl cluster_status
1

03-4、集群的关闭与重启

没有一个直接的命令可以关闭整个集群，需要逐一进行关闭。但是需要保证在重启时，最后关闭的节点最先被启动。如果第一个启动的不是最后关闭的节点，那么这个节点会等待最后关闭的那个节点启动，默认进行 10 次连接尝试，超时时间为 30 秒，如果依然没有等到，则该节点启动失败。

这带来的一个问题是，假设在一个三节点的集群当中，关闭的顺序为 node1，node2，node3，如果 node1 因为故障暂时没法恢复，此时 node2 和 node3 就无法启动。想要解决这个问题，可以先将 node1 节点进行剔除，命令如下：

rabbitmqctl forget_cluster_node rabbit@node1 --offline
1

此时需要加上 -offline 参数，它允许节点在自身没有启动的情况下将其他节点剔除

03-5、集群的解除

两种方式：
1：在当前自己节点重置退出集群
2：在其他任意点操作将某节点踢出集群
注意：被解除集群的节点必须要停止服务，不然无法解除

1、在自己节点重置当前节点，退出集群

# 1.停止服务
rabbitmqctl stop_app
# 2.重置集群状态
rabbitmqctl reset
# 3.重启服务
rabbitmqctl start_app
1
2
3
4
5
6

2、在集群其他任意一个正常节点将节点踢出集群

rabbitmqctl forget_cluster_node rabbit@rabbit-node3
1

03-6、变更节点类型

我们可以将节点的类型从RAM更改为Disk，反之亦然。假设我们想要反转rabbit@rabbit-node2和rabbit@rabbit-node1的类型，将前者从RAM节点转换为磁盘节点，而后者从磁盘节点转换为RAM节点。为此，我们可以使用change_cluster_node_type命令。必须首先停止节点。

# 1.停止服务
rabbitmqctl stop_app
# 2.变更类型 ram disc
rabbitmqctl change_cluster_node_type disc
# 3.重启服务
rabbitmqctl start_app
1
2
3
4
5
6

03-7、清除节点配置

# 如果遇到不能正常退出直接kill进程
systemctl stop rabbitmq-server
# 查看进程
ps aux|grep rabbitmq
# 清楚节点rabbitmq配置
rm -rf /var/lib/rabbitmq/mnesia
1
2
3
4
5
6

04、Dokcer 部署

04-1、docker

单机版：
步骤一：先启动运行三个RabbitMQ容器，这里我选择的是rabbitmq:3.8.3-management，带web插件版本的。
步骤二：加入RabbitMQ节点到集群（备注：参数“–ram”表示设置为内存节点，忽略次参数默认为磁盘节点）

==========================
--link模式网络模式：
==========================
# 启动三个服务
docker run -d --hostname rabbit1 --name myrabbit1 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -p 5672:5672 -p 15672:15672 rabbitmq:3.8.3-management
docker run -d --hostname rabbit2 --name myrabbit2 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -p 5673:5672 -p 15673:15672 --link myrabbit1:rabbit1 rabbitmq:3.8.3-management
docker run -d --hostname rabbit3 --name myrabbit3 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -p 5674:5672 -p 15674:15672 --link myrabbit1:rabbit1 --link myrabbit2:rabbit2 rabbitmq:3.8.3-management


# 设置节点一
docker exec rabbitmq1 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl start_app"

# 设置节点2，加入到集群
docker exec rabbitmq2 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster --ram rabbit@rabbitmq1 && rabbitmqctl start_app"

# 设置节点3，加入到集群：
docker exec rabbitmq3 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster --ram rabbit@rabbitmq1 && rabbitmqctl start_app"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

==========================
network自定义网络模式：
==========================
# 1、先创建自定义网络模式
docker network create rabbit_net
# 2、启动三个RabbitMQ容器，指定network为rabbit_net
docker run -d --hostname rabbitmq1 --name rabbitmq1 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -e RABBITMQ_NODENAME=rabbitmq1 -p 5672:5672 -p 15672:15672 --network rabbit_net rabbitmq:3.8.3-management
docker run -d --hostname rabbitmq2 --name rabbitmq2 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -e RABBITMQ_NODENAME=rabbitmq2 -p 5673:5672 -p 15673:15672 --network rabbit_net rabbitmq:3.8.3-management
docker run -d --hostname rabbitmq3 --name rabbitmq3 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -e RABBITMQ_NODENAME=rabbitmq3 -p 5674:5672 -p 15674:15672 --network rabbit_net rabbitmq:3.8.3-management

# 设置节点一
docker exec rabbitmq1 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl start_app"

# 设置节点2，加入到集群
docker exec rabbitmq2 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster --ram rabbitmq1@rabbitmq1 && rabbitmqctl start_app"

# 设置节点3，加入到集群：
docker exec rabbitmq3 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster --ram rabbitmq1@rabbitmq1 && rabbitmqctl start_app"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

查看集群状态（或者登录web界面查看）

docker exec rabbitmq1 bash -c "rabbitmqctl cluster_status"
1

04-2、docker-compose

步骤一：编写compose.yml配置文件

version: '3.6'
services:
  rabbitmq1:
    image: rabbitmq:3.8.14-management
    container_name: rabbitmq1
    hostname: rabbitmq1
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      - RABBITMQ_ERLANG_COOKIE=CURIOAPPLICATION
      - RABBITMQ_NODENAME:rabbitmq1
    networks:
      - rabbitmq
  rabbitmq2:
    image: rabbitmq:3.8.14-management
    container_name: rabbitmq2
    hostname: rabbitmq2
    ports:
      - "5673:5672"
      - "15673:15672"
    environment:
      - RABBITMQ_ERLANG_COOKIE=CURIOAPPLICATION
      - RABBITMQ_NODENAME:rabbitmq2
    networks:
      - rabbitmq
  rabbitmq3:
    image: rabbitmq:3.8.14-management
    container_name: rabbitmq3
    restart: always
    hostname: rabbitmq3
    ports:
      - "5674:5672"
      - "15674:15672"
    environment:
      - RABBITMQ_ERLANG_COOKIE=CURIOAPPLICATION
      - RABBITMQ_NODENAME:rabbitmq3
    networks:
      - rabbitmq
networks:
  rabbitmq:
    driver: bridge
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

步骤二：设置集群

# 设置节点1
docker exec rabbitmq1 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl start_app"

# 设置节点2，加入到集群
docker exec rabbitmq2 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster --ram rabbitmq1@rabbitmq1 && rabbitmqctl start_app"

# 设置节点3，加入到集群：
docker exec rabbitmq3 bash -c "rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster --ram rabbitmq1@rabbitmq1 && rabbitmqctl start_app"
1
2
3
4
5
6
7
8

05、镜像模式搭建

镜像队列模式相比较普通模式，镜像模式会占用更多的带宽来进行同步，所以镜像队列的吞吐量会低于普通模式

普通模式不能实现高可用，某个节点挂了后，这个节点上的消息将无法被消费，需要等待节点启动后才能被消费

镜像队列是Rabbit2.6.0版本带来的一个新功能，允许内建双活冗余选项，与普通队列不同，镜像节点在集群中的其他节点拥有从队列拷贝，一旦主节点不可用，最老的从队列将被选举为新的主队列。

**镜像队列的工作原理：**在某种程度上你可以将镜像队列视为，拥有一个隐藏的fanout交换器，它指示者信道将消息分发到从队列上。

实现镜像模式非常简单，有3种方式：管理平台命令创建、可视化管理界面手动创建、声明队列的时候创建

1、管理平台命令行创建

# 创建镜像模式
rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]
1
2

下面是对这些配置的解释：

• -p Vhost：可选参数，针对指定vhost下的queue进行设置。
• Name：策略名称。
• Pattern：队列名称的匹配规则，使用正则表达式表示；
• Definition：镜像队列的主体规则，json字符串，包括三个部分：ha-mode, ha-params, ha-sync-mod
  • ha-mode：指明镜像队列的模式，有效值为 all、exactly、nodes
    • all：表示在集群中所有的节点上进行镜像
    • exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
    • nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
  • ha-params：ha-mode模式需要用到的参数补充
  • ha-sync-mode：镜像消息同步方式：automatic（自动）manually（手动-默认）
    • manual手动<默认模式>.新的队列镜像将不会收到现有的消息，它只会接收新的消息。
    • automatic自动同步.当一个新镜像加入时，队列会自动同步。队列同步是一个阻塞操作

创建镜像队列（在任意节点上操作都可以）

# 配置所有名字开头为policy的队列进行镜像，镜像数量为2那么命令如下
rabbitmqctl set_policy -p / ha-policy "^policy_" '{"ha-mode":"exactly","ha-params":2}'

# 设置策略匹配所有名称的队列都进行高可用配置
rabbitmqctl set_policy -p / ha "^" '{"ha-mode":"all","ha-sync-mode":"automatic"}'
1
2
3
4
5

查看镜像队列

# 查看所有的镜像队列
rabbitmqctl list_policies

# 查看vhost下的所有的策略（policies）
rabbitmqctl list_policies -p / 
1
2
3
4
5

删除镜像队列

rabbitmqctl clear_policy [-p <vhost>] <name>

# 清理名称为 ha-policy 的镜像队列
rabbitmqctl clear_policy -p / ha-policy
1
2
3
4

2、可视化管理界面手动创建

添加方式：进入管理页面 -> Admin -> Policies（在页面右侧）-> Add / update a policy

全部节点镜像策略创建：

指定节点数镜像策略创建：

参数说明：

• name: 策略名称，如果使用已有的名称，保存后将会修改原来的信息

• Apply to：策略应用到什么对象上（一般选择交换机和队列一起）

• Pattern：策略应用到对象时，对象名称的匹配规则（正则表达式）

• Priority：优先级，数值越大，优先级越高，相同优先级取最后一个

• Definition：策略定义的类容，对于镜像队列的配置来说，只需要包含3个部分: ha-mode 、ha-params 和 ha-sync-mode。其中，ha-sync-mode是同步的方式，自动还是手动，默认是手动。ha-mode 和 ha-params 组合使用。组合方式如下：

3、声明队列的时候创建（不推荐）

// 声明队列的时候可以加入镜像队列参数（一般不推荐使用）
Map<String,Object> arguments = new HashMap<>();
// x-ha-policy：创建HA队列
arguments.put("x-ha-policy",‘"nodes");
// x-ha-nodes：HA队列的分布节点
arguments.put("x-ha-nodes","rabbit@rabbitmq1");
channel.queueDeclare("ttl-queue", false, false, false, arguments);
1
2
3
4
5
6
7

06、集群监控管理

在广大的互联网行业中RabbitMQ几乎都会有集群，那么对于集群的监控就成了企业生态中必不可少的。主要有如下四种监控：

1、管理界面监控

管理界面监控需要我们开启对应的插件（rabbitmq-plugins enable rabbitmq_management）访问：http://ip:15672

在管理控制台我们就可以直观的看到集群中的每一个节点是否正常,如果为红色则表示节点挂掉了，同时可以很方便的查看到各个节点的内存、磁盘等相关的信息，使用起来也是非常方便的。但是遗憾的该功能做的比较简陋,没有告警等一些列的个性化设置，同时如果想把他接入到公司其他的监控系统统一管理也是很难做到的,所以扩展性不强，一般在小型企业的小集群中使用。

2、Tracing日志监控

对于企业级的应用开发来讲,我们通常都会比较关注我们的消息，甚至很多的场景把消息的可靠性放在第一位，但是我们的MQ集群难免会出现消息异常丢失或者客户端无法发送消息等异常情况，此时为了帮助开发人员快速的定位问题，我们就可以对消息的投递和消费过程进行监控，而Tracing日志监控插件帮我们很好的实现了该功能。

消息中心的消息追踪需要使用Trace实现，Trace是Rabbitmq用于记录每一次发送的消息，方便使用Rabbitmq的开发者调试、排错。可通过插件形式提供可视化界面。Trace启动后会自动创建系统Exchange：amq.rabbitmq.trace，每个队列会自动绑定该Exchange，绑定后发送到队列的消息都会记录到Trace日志。

1、消息追踪启用与查看：以下是trace的相关命令和使用（要使用需要先rabbitmq启用插件，再打开开关才能使用）

安装插件并开启 trace_on 之后，会发现多个 exchange：amq.rabbitmq.trace ，类型为：topic。

2、日志追踪

1、发送消息

rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息--01");
1

2、查看trace

3、点击Tracing查看Trace log files

4、点击xuexiangban-trace.log 确认消息轨迹正确性

3、定制自己的监控系统

RabbitMQ提供了很丰富的restful风格的api接口,我们可以通过这些接口得到对应的集群数据,此时我们就可以定制我们的监控系统。

更多API的相关信息和描述可以访问：http://ip:15672/api/

接下来我们使用RabbitMQ Http API接口来获取集群监控数据：

1、创建一个普通maven项目

2、引入HttpClient以及Jackson的相关Jar依赖

    <properties>
        <!--锁定 jdk 版本为 1.8-->
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.3.6</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.7.4</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-annotations</artifactId>
            <version>2.7.4</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.7.4</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.16</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

3、编写监控代码：

package com.example;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import org.apache.http.HttpEntity;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.auth.BasicScheme;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * RabbitMQ的监控
 */
public class MonitorRabbitMQ {
    // RabbitMQ的HTTP API——获取集群各个实例的状态信息，ip替换为自己部署相应实例的
    private static String RABBIT_NODES_STATUS_REST_URL = "http://127.0.0.1:15672/api/nodes";
    // RabbitMQ的HTTP API——获取集群用户信息，ip替换为自己部署相应实例的
    private static String RABBIT_USERS_REST_URL = "http://127.0.0.1:15672/api/users";
    // rabbitmq的用户名
    private static String RABBIT_USER_NAME = "guest";
    // rabbitmq的密码
    private static String RABBIT_USER_PWD = "guest";

    public static void main(String[] args) {
        try {
            //step1.获取rabbitmq集群各个节点实例的状态信息
            Map<String, ClusterStatus> clusterMap =
                    fetchRabbtMQClusterStatus(RABBIT_NODES_STATUS_REST_URL, RABBIT_USER_NAME, RABBIT_USER_PWD);
            //step2.打印输出各个节点实例的状态信息
            for (Map.Entry entry : clusterMap.entrySet()) {
                System.out.println(entry.getKey() + " : " + entry.getValue());
            }
            //step3.获取rabbitmq集群用户信息
            Map<String, User> userMap =
                    fetchRabbtMQUsers(RABBIT_USERS_REST_URL, RABBIT_USER_NAME, RABBIT_USER_PWD);
            //step4.打印输出rabbitmq集群用户信息
            for (Map.Entry entry : userMap.entrySet()) {
                System.out.println(entry.getKey() + " : " + entry.getValue());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static Map<String, ClusterStatus> fetchRabbtMQClusterStatus(String url, String username, String password) throws IOException {
        Map<String, ClusterStatus> clusterStatusMap = new HashMap<String, ClusterStatus>();
        String nodeData = getData(url, username, password);
        JsonNode jsonNode = null;
        try {
            jsonNode = JsonUtil.toJsonNode(nodeData);
        } catch (IOException e) {
            e.printStackTrace();
        }
        Iterator<JsonNode> iterator = jsonNode.iterator();
        while (iterator.hasNext()) {
            JsonNode next = iterator.next();
            ClusterStatus status = new ClusterStatus();
            status.setDiskFree(next.get("disk_free").asLong());
            status.setFdUsed(next.get("fd_used").asLong());
            status.setMemoryUsed(next.get("mem_used").asLong());
            status.setProcUsed(next.get("proc_used").asLong());
            status.setSocketUsed(next.get("sockets_used").asLong());
            clusterStatusMap.put(next.get("name").asText(), status);
        }
        return clusterStatusMap;
    }
    public static Map<String, User> fetchRabbtMQUsers(String url, String username, String password) throws IOException {
        Map<String, User> userMap = new HashMap<String, User>();
        String nodeData = getData(url, username, password);
        JsonNode jsonNode = null;
        try {
            jsonNode = JsonUtil.toJsonNode(nodeData);
        } catch (IOException e) {
            e.printStackTrace();
        }
        Iterator<JsonNode> iterator = jsonNode.iterator();
        while (iterator.hasNext()) {
            JsonNode next = iterator.next();
            User user = new User();
            user.setName(next.get("name").asText());
            user.setTags(next.get("tags").asText());
            userMap.put(next.get("name").asText(), user);
        }
        return userMap;
    }
    public static String getData(String url, String username, String password) throws IOException {
        CloseableHttpClient httpClient = HttpClients.createDefault();
        UsernamePasswordCredentials creds = new UsernamePasswordCredentials(username, password);
        HttpGet httpGet = new HttpGet(url);
        httpGet.addHeader(BasicScheme.authenticate(creds, "UTF-8", false));
        httpGet.setHeader("Content-Type", "application/json");
        CloseableHttpResponse response = httpClient.execute(httpGet);
        try {
            if (response.getStatusLine().getStatusCode() != 200) {
                System.out.println("call http api to get rabbitmq data return code: " + response.getStatusLine().getStatusCode() + ", url: " + url);
            }
            HttpEntity entity = response.getEntity();
            if (entity != null) {
                return EntityUtils.toString(entity);
            }
        } finally {
            response.close();
        }
        return null;
    }

    public static class JsonUtil {
        private static ObjectMapper objectMapper = new ObjectMapper();
        static {
            objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
            //objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
        }
        public static JsonNode toJsonNode(String jsonString) throws IOException {
            return objectMapper.readTree(jsonString);
        }
    }

    @Data
    public static class User {
        private String name;
        private String tags;
    }

    @Data
    public static class ClusterStatus {
        private long diskFree;
        private long diskLimit;
        private long fdUsed;
        private long fdTotal;
        private long socketUsed;
        private long socketTotal;
        private long memoryUsed;
        private long memoryLimit;
        private long procUsed;
        private long procTotal;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

控制台输出：

rabbit-3@CentOS7 : MonitorRabbitMQ.ClusterStatus(diskFree=36523638784, diskLimit=0, fdUsed=34, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=86605824, memoryLimit=0, procUsed=443, procTotal=0)
rabbit-1@CentOS7 : MonitorRabbitMQ.ClusterStatus(diskFree=36523638784, diskLimit=0, fdUsed=36, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=89972736, memoryLimit=0, procUsed=444, procTotal=0)
rabbit-2@CentOS7 : MonitorRabbitMQ.ClusterStatus(diskFree=36523638784, diskLimit=0, fdUsed=34, fdTotal=0, socketUsed=0, socketTotal=0, memoryUsed=89382912, memoryLimit=0, procUsed=442, procTotal=0)
admin : MonitorRabbitMQ.User(name=admin, tags=administrator)
guest : MonitorRabbitMQ.User(name=guest, tags=administrator)
1
2
3
4
5

4、Zabbix监控RabbitMQ

Zabbix是一个基于WEB界面提供分布式系统监视以及网络监视功能的企业级开源解决方案,他也可以帮助我们搭建一个MQ集群的监控系统,同时提供预警等功能，但是由于其搭建配置要求比较高一般都是由运维人员负责搭建,感兴趣的同学可以访问：https://www.zabbix.com/ 官网进行了解学习。

07、SpringBoot集群配置

1、引入相关依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.6.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9
10

2、配置文件详细

 rabbitmq:
    # 指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host)
    addresses: 127.0.0.1:6605,127.0.0.1:6606,127.0.0.1:6705 
#    port:
    ##集群配置 addresses之间用逗号隔开
    # addresses: ip:port,ip:port
    password: admin
    username: 123456
    virtual-host: / # 连接到rabbitMQ的vhost
    requested-heartbeat: #指定心跳超时，单位秒，0为不指定；默认60s
    publisher-confirms: #是否启用 发布确认
    publisher-reurns: # 是否启用发布返回
    connection-timeout: #连接超时，单位毫秒，0表示无穷大，不超时
    cache:
      channel.size: # 缓存中保持的channel数量
      channel.checkout-timeout: # 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel
      connection.size: # 缓存的连接数，只有是CONNECTION模式时生效
      connection.mode: # 连接工厂缓存模式：CHANNEL 和 CONNECTION
    listener:
      simple.auto-startup: # 是否启动时自动启动容器
      simple.acknowledge-mode: # 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto
      simple.concurrency: # 最小的消费者数量
      simple.max-concurrency: # 最大的消费者数量
      simple.prefetch: # 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量.
      simple.transaction-size: # 指定一个事务处理的消息数量，最好是小于等于prefetch的数量.
      simple.default-requeue-rejected: # 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
      simple.idle-event-interval: # 多少长时间发布空闲容器时间，单位毫秒
      simple.retry.enabled: # 监听重试是否可用
      simple.retry.max-attempts: # 最大重试次数
      simple.retry.initial-interval: # 第一次和第二次尝试发布或传递消息之间的间隔
      simple.retry.multiplier: # 应用于上一重试间隔的乘数
      simple.retry.max-interval: # 最大重试时间间隔
      simple.retry.stateless: # 重试是有状态or无状态
    template:
      mandatory: # 启用强制信息；默认false
      receive-timeout: # receive() 操作的超时时间
      reply-timeout: # sendAndReceive() 操作的超时时间
      retry.enabled: # 发送重试是否可用
      retry.max-attempts: # 最大重试次数
      retry.initial-interval: # 第一次和第二次尝试发布或传递消息之间的间隔
      retry.multiplier: # 应用于上一重试间隔的乘数
      retry.max-interval: #最大重试时间间隔
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

注：相关配置很多，大家只需要关注一些常用的配置即可

对于发送方而言，需要做以下配置：

• 1、配置CachingConnectionFactory

• 2、配置Exchange/Queue/Binding

• 3、配置RabbitAdmin创建上一步的Exchange/Queue/Binding

• 4、配置RabbitTemplate用于发送消息，RabbitTemplate通过CachingConnectionFactory获取到Connection，然后想指定Exchange发送

对于消费方而言，需要做以下配置：

• 1、配置CachingConnectionFactory
• 2、配置Exchange/Queue/Binding
• 3、配置RabbitAdmin创建上一步的Exchange/Queue/Binding
• 4、配置RabbitListenerContainerFactory
• 5、配置@RabbitListener/@RabbitHandler用于接收消息

在默认情况下主要的配置如下：

3、Spring-AMQP主要对象

注：如果不了解AMQP请前往官网了解.

4、通过配置类加载的方式

package com.yd.demo.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitConfig {
    private static final Logger logger = LoggerFactory.getLogger(RabbitConfig.class);
    public static final String RECEIVEDLXEXCHANGE="spring-ex";
    public static final String RECEIVEDLXQUEUE="spring-qu1";
    public static final String RECEIVEDLXROUTINGKEY="aa";
    public static final String DIRECTEXCHANGE="spring-ex";
    public static final String MDMQUEUE="mdmQueue";
    public static final String TOPICEXCHANGE="spring-top";
    @Value("${spring.rabbitmq.addresses}")
    private String hosts;
    @Value("${spring.rabbitmq.username}")
    private String userName;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;
 /*   @Value("${rabbit.channelCacheSize}")
    private int channelCacheSize;*/
//    @Value("${rabbit.port}")
//    private int port;
/*    @Autowired
    private ConfirmCallBackListener confirmCallBackListener;
    @Autowired
    private ReturnCallBackListener returnCallBackListener;*/
    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setAddresses(hosts);
        cachingConnectionFactory.setUsername(userName);
        cachingConnectionFactory.setPassword(password);
//        cachingConnectionFactory.setChannelCacheSize(channelCacheSize);
        //cachingConnectionFactory.setPort(port);
        cachingConnectionFactory.setVirtualHost(virtualHost);
        //设置连接工厂缓存模式：
        cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        //缓存连接数
        cachingConnectionFactory.setConnectionCacheSize(3);
        //设置连接限制
        cachingConnectionFactory.setConnectionLimit(6);
        logger.info("连接工厂设置完成，连接地址{}"+hosts);
        logger.info("连接工厂设置完成，连接用户{}"+userName);
        return cachingConnectionFactory;
    }
    @Bean
    public RabbitAdmin rabbitAdmin(){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setAutoStartup(true);
        rabbitAdmin.setIgnoreDeclarationExceptions(true);
        rabbitAdmin.declareBinding(bindingMdmQueue());
        //声明topic交换器
        rabbitAdmin.declareExchange(directExchange());
        logger.info("管理员设置完成");
        return rabbitAdmin;
    }
    @Bean
    public RabbitListenerContainerFactory listenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //最小消费者数量
        factory.setConcurrentConsumers(10);
        //最大消费者数量
        factory.setMaxConcurrentConsumers(10);
        //一个请求最大处理的消息数量
        factory.setPrefetchCount(10);
        //
        factory.setChannelTransacted(true);
        //默认不排队
        factory.setDefaultRequeueRejected(true);
        //手动确认接收到了消息
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        logger.info("监听者设置完成");
        return factory;
    }
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange(DIRECTEXCHANGE,true,false);
    }
    @Bean
    public Queue mdmQueue(){
        Map arguments = new HashMap<>();
        // 绑定该队列到私信交换机
        arguments.put("x-dead-letter-exchange",RECEIVEDLXEXCHANGE);
        arguments.put("x-dead-letter-routing-key",RECEIVEDLXROUTINGKEY);
        logger.info("队列交换机绑定完成");
        return new Queue(RECEIVEDLXQUEUE,true,false,false,arguments);
    }
    @Bean
    Binding bindingMdmQueue() {
        return BindingBuilder.bind(mdmQueue()).to(directExchange()).with("");
    }
    @Bean
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        rabbitTemplate.setMandatory(true);
        //发布确认
//        rabbitTemplate.setConfirmCallback(confirmCallBackListener);
        // 启用发布返回
//        rabbitTemplate.setReturnCallback(returnCallBackListener);
        logger.info("连接模板设置完成");
        return rabbitTemplate;
    }
  /*  @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(TOPICEXCHANGE,true,false);
    }*/
  /*
*//**
     * @return DirectExchange
     *//*
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(RECEIVEDLXEXCHANGE,true,false);
    }
*//*
*
     * @return Queue
*//*
    @Bean
    public Queue dlxQueue() {
        return new Queue(RECEIVEDLXQUEUE,true);
    }
*//*
     * @return Binding
     *//*
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(RECEIVEDLXROUTINGKEY);
    }*/
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

5、总结

通过两种方式加载：1.通过配置文件、 2.通过配置类

说明：上面是通过配置文件与配置类的方式去加载,常用的配置如上所示。实际使用中要生产方与消费方要分开配置，相关配置也会有小变动，大体配置不变。更多信息可查看官网配置。