1.环境安装

1.1 采用docker方式安装

命令如下：

docker run -d --hostname ecs-01-0003 --name rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -e RABBITMQ_DEFAULT_VHOST=middle -p 15672:15672 -p 5672:5672 -v /data/rabbitmq:/var/lib/rabbitmq rabbitmq:3.7.28-management

1.2 windows环境下安装

windows下安装需要安装erlang包，然后配置环境变量，在安装配置即可，这种方式比较简单，略过

参看：https://www.cnblogs.com/chenwolong/p/rabbitmq.html

1.3 Linux环境下安装（CentOs7.x）

1.3.1 安装依赖环境

在 http://www.rabbitmq.com/which-erlang.html 页面查看安装rabbitmq需要安装erlang对应的版本
在 https://github.com/rabbitmq/erlang-rpm/releases 页面找到需要下载的erlang版本，erlang-*.centos.x86_64.rpm 就是centos版本的。

复制下载地址后，使用wget命令下载

wget -P /home/download https://github.com/rabbitmq/erlang- rpm/releases/download/v23.3.3/erlang-23.3.3-1.el7.x86_64.rpm

安装 Erlang

sudo rpm -Uvh /home/download/erlang-21.2.3-1.el7.centos.x86_64.rpm

安装 socat
```
sudo yum install -y socat
```

1.3.2 安装RabbitMq

在官方下载页面找到CentOS7版本的下载链接，下载rpm安装包
```
wget -P /home/download https://github.com/rabbitmq/rabbitmq- server/releases/download/v3.8.15/rabbitmq-server-3.8.15-1.el7.noarch.rpm
```
提示：可以在 https://github.com/rabbitmq/rabbitmq-server/tags 下载历史版本

安装RabbitMQ

sudo rpm -Uvh /home/download/rabbitmq-server-3.7.9-1.el7.noarch.rpm

1.3.3 启动和关闭

启动服务
```
sudo systemctl start rabbitmq-server
```
查看状态
```
sudo systemctl status rabbitmq-server
```
停止服务
```
sudo systemctl stop rabbitmq-server
```
设置开机启动
```
sudo systemctl enable rabbitmq-server
```

1.3.4 开启web管理插件

开启插件

rabbitmq-plugins enable rabbitmq_management

说明：rabbitmq有一个默认的guest用户，但只能通过localhost访问，所以需要添加一个能够远程访问的用户。

添加用户
```
rabbitmqctl add_user admin admin
```

为用户分配操作权限

rabbitmqctl set_user_tags admin administrator

为用户分配资源权限

rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

1.3.5 防火墙端口配置

RabbitMQ 服务启动后，还不能进行外部通信，需要将端口添加都防火墙

添加端口

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.3.6 RabbitMQ基本配置

RabbitMQ 有一套默认的配置，能够满足日常开发需求，如果需要修改，需要自己创建一个配置文件

touch /etc/rabbitmq/rabbitmq.conf

配置文件示例：

https://github.com/rabbitmq/rabbitmq-server/blob/master/deps/rabbit/docs/rabbitmq.conf.example

配置项说明：

https://www.rabbitmq.com/configure.html#config-items

RabbitMQ 会绑定一些端口，安装完后，需要将这些端口添加至防火墙。

4369：是Erlang的端口/结点名称映射程序，用来跟踪节点名称监听地址，在集群中起到一个类似

DNS的作用。

5672, 5671：AMQP 0-9-1 和 1.0 客户端端口，没有使用SSL和使用SSL的端口。
25672：用于RabbitMQ节点间和CLI工具通信，配合4369使用。
15672：HTTP_API端口，管理员用户才能访问，用于管理RbbitMQ，需要启用management插

件。

61613,61614：当STOMP插件启用的时候打开，作为STOMP客户端端口（根据是否使用TLS选

择）。

1883, 8883：当MQTT插件启用的时候打开，作为MQTT客户端端口（根据是否使用TLS选择）。
15674：基于WebSocket的STOMP客户端端口（当插件Web STOMP启用的时候打开）
15675：基于WebSocket的MQTT客户端端口（当插件Web MQTT启用的时候打开）

SpringBoot集成rabbitmq错误:org.springframework.amqp.AmqpConnectException: java.net.ConnectException的解决办法

进入到rabbitmq的目录并且在此目录下编辑一个名为rabbitmq.config的文件(注意:名字一定要是这个)

在这个文件中加入如下代码保存之后重启RabbitMq，再次启动Java项目即可正常的访问了!

[{rabbit, [{loopback_users, []}]}].

1.3.7 RabbitMQ管理界面

RabbitMQ 安装包中带有管理插件，但需要手动激活

rabbitmq-plugins enable rabbitmq_management

RabbitMQ 有一个默认的用户“guest”，但这个用户默认只能通过本机访问，要让其它机器可以访问，需要创建一个新用户，为其分配权限

#添加用户 
rabbitmqctl add_user admin admin 
#为用户分配权限 
rabbitmqctl set_user_tags admin administrator 
#为用户分配资源权限 
rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

1.3.8 RabbitMQ角色分类

RabbitMQ 的用户角色分类：none、management、policymaker、monitoring、administrator

none：不能访问 management plugin
management：用户可以通过AMQP做的任何事外加：

列出自己可以通过AMQP登入的virtual hosts

查看自己的virtual hosts中的queues, exchanges 和 bindings

查看和关闭自己的channels 和 connections

查看有关自己的virtual hosts的“全局”的统计信息，包含其他用户在这些virtual hosts中的活动。
policymaker：management可以做的任何事外加：

查看、创建和删除自己的virtual hosts所属的policies和parameters
monitoring：management可以做的任何事外加：

列出所有virtual hosts，包括他们不能登录的virtual hosts

查看其他用户的connections和channels

查看节点级别的数据如clustering和memory使用情况

查看真正的关于所有virtual hosts的全局的统计信息
administrator：policymaker和monitoring可以做的任何事外加：

创建和删除virtual hosts

查看、创建和删除users

查看创建和删除permissions

关闭其他用户的connections

1.4 多机多节点集群方式部署

1.4.1 环境准备

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

很多磁盘空间

1.4.2 修改配置文件
```
sudo scp /etc/hosts root@node2:/etc/ 
sudo scp /etc/hosts root@node3:/etc/
```
说明：命令中的root是目标机器的用户名，命令执行后，可能会提示需要输入密码，输入对应用户的密码

就行了

提示：如果是通过克隆的VM，可以省略这一步

1.4.3 防火墙端口配置

1.4.4 启动RabbitMq

准备三台安装好RabbitMQ 的机器，安装方法见安装步骤

10.10.1.41

10.10.1.42

10.10.1.43

添加端口

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

将 10.10.1.41 上的 /var/lib/rabbitmq/.erlang.cookie 文件复制到另外两台机器上

scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/ 
scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/

修改 10.10.1.41 机器上的 /etc/hosts 文件
```
sudo vim /etc/hosts
```

添加IP和节点名

10.10.1.41 node1 
10.10.1.42 node2 
10.10.1.43 node3

修改对应主机的hostname

hostname node1 
hostname node2 
hostname node3

将 10.10.1.41 上的hosts文件复制到另外两台机器上

启动每台机器的RabbitMQ

sudo systemctl start rabbitmq-server

或者

rabbitmq-server -detached

将 10.10.1.42 加入到集群

# 停止RabbitMQ 应用 
rabbitmqctl stop_app 
# 重置RabbitMQ 设置 
rabbitmqctl reset 
# 加入到集群 
rabbitmqctl join_cluster rabbit@node1 --ram 
# 启动RabbitMQ 应用 
rabbitmqctl start_app

查看集群状态，看到 running_nodes,[rabbit@node1,rabbit@node2] 表示节点启动成功
```
rabbitmqctl cluster_status
```

将 10.10.1.43 加入到集群

# 停止 RabbitMQ 应用 
rabbitmqctl stop_app 
# 重置 RabbitMQ 设置 
rabbitmqctl reset 
# 节点加入到集群 
rabbitmqctl join_cluster rabbit@node1 --ram 
# 启动 RabbitMQ 应用 
rabbitmqctl start_app

重复地3步，查看集群状态

1.5 单机多节点部署

环境准备

准备一台已经安装好RabbitMQ 的机器，安装方法见上面 Linux环境下安装

10.10.1.41

2.防火墙端口配置

需要将每个节点的端口都添加到防火墙

添加端口

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 --zone=public --add-port=5673/tcp --permanent 
sudo firewall-cmd --zone=public --add-port=25673/tcp --permanent 
sudo firewall-cmd --zone=public --add-port=15673/tcp --permanent 
sudo firewall-cmd --zone=public --add-port=5674/tcp --permanent 
sudo firewall-cmd --zone=public --add-port=25674/tcp --permanent 
sudo firewall-cmd --zone=public --add-port=15674/tcp --permanent

重启防火墙
```
sudo firewall-cmd --reload
```

3.启动RabbitMq

在启动前，先修改RabbitMQ 的默认节点名（非必要），在 /etc/rabbitmq/rabbitmq-env.conf

增加以下内容

# RabbitMQ 默认节点名，默认是
rabbit RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit1

RabbitMQ 默认是使用服务的启动的，单机多节点时需要改为手动启动，先停止运行中的 RabbitMQ 服务
```
sudo systemctl stop rabbitmq-server
```
启动第一个节点
```
rabbitmq-server -detached
```

启动第二个节点

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

将rabbit2加入到集群

# 停止 rabbit2 的应用 
rabbitmqctl -n rabbit2 stop_app 
# 重置 rabbit2 的设置 
rabbitmqctl -n rabbit2 reset 
# rabbit2 节点加入到 rabbit1的集群中 
rabbitmqctl -n rabbit2 join_cluster rabbit1 --ram 
# 启动 rabbit2 节点
rabbitmqctl -n rabbit2 start_app

将rabbit3加入到集群

# 停止 rabbit3 的应用 
rabbitmqctl -n rabbit3 stop_app 
# 重置 rabbit3 的设置
rabbitmqctl -n rabbit3 reset 
# rabbit3 节点加入到 rabbit1的集群中 
rabbitmqctl -n rabbit3 join_cluster rabbit1 --ram 
# 启动 rabbit3 节点 rabbitmqctl -n rabbit3 start_app

查看集群状态，看到 {running_nodes,[rabbit3@node1,rabbit2@node1,rabbit1@node1]} 说明节点已启动成功。
```
rabbitmqctl cluster_status
```
提示：在管理界面可以更直观的看到集群信息

移除节点

# 首先将要移除的节点停机 
rabbitmqctl -n rabbit3 stop_app 
# 在主节点,也就是发起进群的主机上进行节点的移除 
rabbitmqctl -n rabbit forget_cluster_node rabbit3 1234

1.6 镜像队列模式集群部署

镜像队列属于RabbitMQ 的高可用方案，见：https://www.rabbitmq.com/ha.html#mirroring-arguments
通过前面的步骤搭建的集群属于普通模式集群，是通过共享元数据实现集群
开启镜像队列模式需要在管理页面添加策略，添加方式：

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

在表单中填入：

name: ha-all 
Pattern: ^ 
Apply to: Queues 
Priority: 0 
Definition: ha-mode = all

参数说明

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

Apply to：策略应用到什么对象上

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

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

Definition：策略定义的类容，对于镜像队列的配置来说，只需要包含3个部分: ha-mode 、

ha-params 和 ha-sync-mode 。其中， ha-sync-mode 是同步的方式，自动还是手动，默认是自动。

ha-mode 和 ha-params 组合使用。

组合方式如下：

镜像队列模式相比较普通模式，镜像模式会占用更多的带宽来进行同步，所以镜像队列的吞吐量会低于普通模式
但普通模式不能实现高可用，某个节点挂了后，这个节点上的消息将无法被消费，需要等待节点启动后才能被消费。

2.延迟插件安装

2.1 插件官网下载地址

插件下载地址：https://www.rabbitmq.com/community-plugins.html
找到相应的版本rabbitmq_delayed_message_exchange下载，链接如下：
插件gitHub下载页面:
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

2.2 docker容器内安装延迟插件

拷贝至docker容器内
docker cp rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq容器ID:/plugins
进入docker容器内
docker exec -it rabbitmq bash
赋予权限
chmod 777 /plugins/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez
启动延时插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

3.RabbitMq的概念模型

3.1 什么是消息队列？

简单来说就是应用之间传递的数据通过一个中转站来做中转，是应用之间的一种通信方式和数据交互的方式。

3.2 为什么要用消息队列？

削峰流控、异步解耦（同步接口调用改为业务异步解耦）、最终一致性、广播、延迟队列等

3.3 RabbitMq的特点？

AMQP ：Advanced Message Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特点包括，两种说法总结如下：

3.3.1 特点一总结

3.3.1.1 可靠性（Reliability）

RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。

3.3.1.2 灵活的路由（Flexible Routing）

在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供

了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也

通过插件机制实现自己的 Exchange 。

3.3.1.3 消息集群（Clustering）

多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。

3.3.1.4 高可用（Highly Available Queues）

队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。

3.3.1.5 多种协议（Multi-protocol）

RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。

3.3.1.6 多语言客户端（Many Clients）

RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。

3.3.1.7 管理界面（Management UI）

RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。

3.3.1.8 跟踪机制（Tracing）

如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。

3.3.1.9 插件机制（Plugin System）

RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

3.3.2 特点二总结

3.3.2.1 RabbitMQ底层使用Erlang语言编写，传递效率高，延迟低

3.3.2.2 开源、性能优秀、稳定性较高

3.3.2.3 与SpringAMQP完美的整合、API丰富

3.3.2.4 集群模式丰富、表达式配置、HA模式、镜像队列模式

3.3.2.5 保证数据不丢失的情况下，做到高可用

3.3.2.6 AMQP全称：Advanced Message Queuing Protocol

3.3.2.7 AMQP翻译:高级消息队列协议

3.4 消息模型

3.5 RabbitMQ 内部结构

3.5.1 Message

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、 delivery-mode（指出该消息可能需要持久性存储）等。

3.5.2 Publisher

消息的生产者，也是一个向交换器发布消息的客户端应用程序。

3.5.3 Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

3.5.4 Binding

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

3.5.5 Queue

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

3.5.6 Connection

网络连接，比如一个TCP连接。

3.5.7 Channel

信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。AMQP 的消息路由过程

3.5.8 Consumer

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

3.5.9 Virtual Host

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost

是 / 。

3.5.10 Broker

表示消息队列服务器实体。

3.6 消息路由

3.6.1 AMQP 的消息路由过程

3.7 交换机类型

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型：

3.7.1 direct

消息中的路由键（routing key）如果和 Binding 中的 binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard”等等。它是完全匹配、单播的模式。

3.7.2 fanout

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

3.7.3 topic

topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个

通配符：符号“#”和符号“”。#匹配0个或多个单词，匹配不多不少一个单词。

3.8 RabbitMQ提供了7 种模式

RabbitMQ提供了7 种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式，发布确认模式

官网对应模式介绍：

https://www.rabbitmq.com/getstarted.html

4.什么是延迟队列？以及它的应用场景？

延时队列，首先，它是一种队列，队列意味着内部的元素是有序的，元素出队和入队是有方向性的，元素从一端进入，从另一端取出。

其次，延时队列，最重要的特性就体现在它的延时属性上，跟普通的队列不一样的是，普通队列中的元素总是等着希望被早点取出处理，而延时队列中的元素则是希望被在指定时间得到取出和处理，所以延时队列中的元素是都是带时间属性的，通常来说是需要被处理的消息或者任务。

简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

那么什么时候需要用延时队列呢？考虑一下以下场景：

订单在十分钟之内未支付则自动取消。
新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
账单在一周内未支付，则自动结算。
用户注册成功后，如果三天内没有登陆则进行短信提醒。
用户发起退款，如果三天内没有得到处理则通知相关运营人员。
预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。

5.项目集成

5.1 pom依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

生产者和消费者端都要引入该依赖

5.2 yml配置

server:
  port: 8080
spring:
  rabbitmq:
    addresses: xxxxxx
    port: 5672
    username: admin
    password: admin
     #虚拟host 可以不设置,使用server默认host / 
    virtual-host: /xxxxx

6.错误姿势

DelayMessageSender如下：

package xxxxxxxx;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

/**
 * @author zlf
 * @description:
 * @time: 2021/7/12 9:51
 */
@Component
@RefreshScope
public class DelayMessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 60秒
     */
    @Value("${delayTime:6000}")
    private Integer delayTime;
    /**
     * 延迟队列发送任意延迟时间的方法
     * @param msg
     * @param delayedExchangeName
     * @param delayedRoutingKey
     */
    public void sendDelayMsg1(Object msg,String delayedExchangeName,String delayedRoutingKey) {
        rabbitTemplate.convertAndSend(delayedExchangeName, delayedRoutingKey, msg, a ->{
            // 每条消息设置一个过期时间
            a.getMessageProperties().setDelay(this.delayTime);
            return a;
        });
    }

}

RabbitMqConfig如下：

package xxxxxxx;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @author zlf
 * @description:
 * @time: 2021/7/12 10:00
 */
@Configuration
public class RabbitMqConfig {

    public static final String DELAY_EXCHANGE_NAME = "delay.queue.xxx.exchange";
    public static final String DELAY_QUEUEC_NAME = "delay.queue.xxx.queuec";
    public static final String DELAY_QUEUEC_ROUTING_KEY ="delay.queue.xxx.routingkey";
    
    /**
     * 声明延时Exchange
     *
     * @return
     */
    @Bean("delayExchange")
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>(2);
        args.put("x-delayed-type", "direct");
        // 下面的这个交换机的姿势有问题,会导致消息发送时而可以收到时而丢失的诡异问题
        DirectExchange directExchange = new DirectExchange(DELAY_EXCHANGE_NAME, true, false, args);
        directExchange.setDelayed(true);
        return directExchange;
    }

    /**
     * 声明延时队列C
     *
     * @return
     */
    @Bean("delayQueueC")
    public Queue delayQueueC() {
        return QueueBuilder.durable(DELAY_QUEUEC_NAME).build();
    }

    /**
     * 声明队列C绑定关系
     *
     * @return
     */
    @Bean
    public Binding deadLetterBindingC() {
        return new Binding(DELAY_QUEUEC_NAME, Binding.DestinationType.QUEUE, DELAY_EXCHANGE_NAME, DELAY_QUEUEC_ROUTING_KEY, null);
    }

}

由于之前用RabbitMq实现一个延迟队列的功能，由于姿势不对导致消息时而有时而丢失的问题，上面的代码其实是一种插件和私信队列混用导致的，由于我们的环境是使用了延迟插件的方式，所以上面这种写法就会翻车了，这个翻车让我排查了一个多星期。

7.正确姿势

正确姿势有两种：延迟插件和死信队列

7.1 延迟插件实现

生产者端的代码：

DelayMessageSender跟上面的是一样的，请看上面6的DelayMessageSender。

RabbitMqConfig如下：

package xxxxxxx;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @author zlf
 * @description:
 * @time: 2021/7/12 10:00
 */
@Configuration
public class RabbitMqConfig {

    public static final String DELAY_EXCHANGE_NAME = "delay.queue.xxx.exchange";
    public static final String DELAY_QUEUEC_NAME = "delay.queue.xxx.queuec";
    public static final String DELAY_QUEUEC_ROUTING_KEY ="delay.queue.xxx.routingkey";

    /**
     * 声明延时Exchange
     *
     * @return
     */
    @Bean("delayExchange")
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>(2);
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    /**
     * 声明延时队列C
     *
     * @return
     */
    @Bean("delayQueueC")
    public Queue delayQueueC() {
        return QueueBuilder.durable(DELAY_QUEUEC_NAME).build();
    }

    /**
     * 声明队列C绑定关系
     *
     * @return
     */
    @Bean
    public Binding deadLetterBindingC() {
        return new Binding(DELAY_QUEUEC_NAME, Binding.DestinationType.QUEUE, DELAY_EXCHANGE_NAME, DELAY_QUEUEC_ROUTING_KEY, null);
    }

}

消费中端的代码：

/**
 * @author zlf
 * @description:
 * @time: 2021/7/12 10:09
 * MQ监听消费
 */
@Slf4j
@Component
@Transactional(rollbackFor = Exception.class)
public class MyConsumer {

    private static final String DELAY_QUEUEC_NAME = "delay.queue.xxxx";

    @RabbitHandler
    @RabbitListener(queues = DELAY_QUEUEC_NAME)
    public void receiveC(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody(), "UTF-8");
        if (StringUtils.isEmpty(msg)) {
            log.info("=========message为空！==============");
            return;
        }
        // 消费消息 TODO 业务处理
        ..............
            ...............
        //关闭手动确认
        //channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        DefaultConsumer deliverCallback = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("=============消息消费成功============= message: {}", new String(body, "UTF-8"));
                //手工确认，第一个参数是消息的id,第二个参数是批量标识（ture标识批量）
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        //消费自动确认，一旦选择这种模式，消费发送到消费端，不管消息是否被消费端正常消费，都会将队列中的消息删除掉
        //channel.basicConsume(DELAY_QUEUEC_NAME,true, deliverCallback);
        //手工确认
        //channel.basicConsume(DELAY_QUEUEC_NAME,false, deliverCallback);
    }
}

7.2 死信队列实现延迟队列

7.2.1 什么是死信队列？

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX，说白了就是一个备胎

7.2.2 消息成为死信的三种情况

队列消息长度到达限制；

消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；

pom配置：

spring:
  rabbitmq:
    host: localhost
    password: guest
    username: guest
    listener:
      type: simple
      simple:
          default-requeue-rejected: false # 消息拒绝策略
          acknowledge-mode: manual # 消费者手动ack确认

原队列存在消息过期设置，消息到达超时时间未被消费；

7.2.3 队列绑定死信交换机

给队列设置参数：x-dead-letter-exchange 和 x-dead-letter-routing-key

也就是说此时Queue作为"生产者"

7.2.4 TTL+死信队列组合实现延迟队列

实现举例如下：

实现示例如下：

DelayMessageSender跟上面的是一样的，请看上面6的DelayMessageSender。

RabbitMQConfig 如下：

@Configuration
public class RabbitMQConfig {

    public static final String DELAY_EXCHANGE_NAME = "delay.queue.demo.business.exchange";
    public static final String DELAY_QUEUEC_NAME = "delay.queue.demo.business.queuec";
    public static final String DELAY_QUEUEC_ROUTING_KEY = "delay.queue.demo.business.queuec.routingkey";
    public static final String DEAD_LETTER_EXCHANGE = "delay.queue.demo.deadletter.exchange";
    public static final String DEAD_LETTER_QUEUEC_ROUTING_KEY = "delay.queue.demo.deadletter.delay_anytime.routingkey";
    public static final String DEAD_LETTER_QUEUEC_NAME = "delay.queue.demo.deadletter.queuec";

    // 声明延时Exchange
    @Bean("delayExchange")
    public DirectExchange delayExchange(){
        return new DirectExchange(DELAY_EXCHANGE_NAME);
    }

    // 声明死信Exchange
    @Bean("deadLetterExchange")
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(DEAD_LETTER_EXCHANGE);
    }

    // 声明延时队列C 不设置TTL
    // 并绑定到对应的死信交换机
    @Bean("delayQueueC")
    public Queue delayQueueC(){
        Map<String, Object> args = new HashMap<>(3);
        // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        // x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEC_ROUTING_KEY);
        return QueueBuilder.durable(DELAY_QUEUEC_NAME).withArguments(args).build();
    }

    // 声明死信队列C 用于接收延时任意时长处理的消息
    @Bean("deadLetterQueueC")
    public Queue deadLetterQueueC(){
        return new Queue(DEAD_LETTER_QUEUEC_NAME);
    }

    // 声明延时列C绑定关系
    @Bean
    public Binding delayBindingC(@Qualifier("delayQueueC") Queue queue,
                                 @Qualifier("delayExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEC_ROUTING_KEY);
    }

    // 声明死信队列C绑定关系
    @Bean
    public Binding deadLetterBindingC(@Qualifier("deadLetterQueueC") Queue queue,
                                      @Qualifier("deadLetterExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEC_ROUTING_KEY);
    }
}

增加一个死信队列C的消费者：

@RabbitListener(queues = DEAD_LETTER_QUEUEC_NAME)
public void receiveC(Message message, Channel channel) throws IOException {
    String msg = new String(message.getBody());
    log.info("当前时间：{},死信队列C收到消息：{}", new Date().toString(), msg);
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

上面是死信对列实现延迟对列的实现

7.3 死信对列的使用

下面是关于死信对列的使用

7.3.1 消息成为死信的三种情况

死信，在官网中对应的单词为“Dead Letter”，可以看出翻译确实非常的简单粗暴。那么死信是个什么东西呢？“死信”是RabbitMQ中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况：

消息被否定确认，使用 channel.basicNack 或 channel.basicReject ，并且此时requeue 属性被设置为false。

pom配置：

spring:
  rabbitmq:
    host: localhost
    password: guest
    username: guest
    listener:
      type: simple
      simple:
          default-requeue-rejected: false # 消息拒绝策略
          acknowledge-mode: manual # 消费者手动ack确认

消息在队列的存活时间超过设置的TTL时间。
消息队列的消息数量已经超过最大队列长度。

那么该消息将成为“死信”。“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃。

7.3.2 如何配置死信队列呢？

其实很简单，大概可以分为以下步骤：

配置业务队列，绑定到业务交换机上
为业务队列配置死信交换机和路由key
为死信交换机配置死信队列

注意: 并不是直接声明一个公共的死信队列，然后所以死信消息就自己跑到死信队列里去了。而是为每个需要使用死信的业务队列配置一个死信交换机，这里同一个项目的死信交换机可以共用一个，然后为每个业务队列分配一个单独的路由key。

有了死信交换机和路由key后，接下来，就像配置业务队列一样，配置死信队列，然后绑定在死信交换机上。也就是说，死信队列并不是什么特殊的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机，只不过是用来接受死信的交换机，所以可以为任何类型【Direct、Fanout、Topic】。一般来说，会为每个业务队列分配一个独有的路由key，并对应的配置一个死信队列进行监听，也就是说，一般会为每个重要的业务队列配置一个死信队列。

7.3.3 死信对列没有延迟的使用

RabbitMQConfig如下：

@Configuration
public class RabbitMQConfig {

    public static final String BUSINESS_EXCHANGE_NAME = "dead.letter.demo.simple.business.exchange";
    public static final String BUSINESS_QUEUEA_NAME = "dead.letter.demo.simple.business.queuea";
    public static final String BUSINESS_QUEUEB_NAME = "dead.letter.demo.simple.business.queueb";
    public static final String DEAD_LETTER_EXCHANGE = "dead.letter.demo.simple.deadletter.exchange";
    public static final String DEAD_LETTER_QUEUEA_ROUTING_KEY = "dead.letter.demo.simple.deadletter.queuea.routingkey";
    public static final String DEAD_LETTER_QUEUEB_ROUTING_KEY = "dead.letter.demo.simple.deadletter.queueb.routingkey";
    public static final String DEAD_LETTER_QUEUEA_NAME = "dead.letter.demo.simple.deadletter.queuea";
    public static final String DEAD_LETTER_QUEUEB_NAME = "dead.letter.demo.simple.deadletter.queueb";

    // 声明业务Exchange
    @Bean("businessExchange")
    public FanoutExchange businessExchange(){
        return new FanoutExchange(BUSINESS_EXCHANGE_NAME);
    }

    // 声明死信Exchange
    @Bean("deadLetterExchange")
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(DEAD_LETTER_EXCHANGE);
    }

    // 声明业务队列A
    @Bean("businessQueueA")
    public Queue businessQueueA(){
        Map<String, Object> args = new HashMap<>(2);
//       x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
//       x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEA_ROUTING_KEY);
        return QueueBuilder.durable(BUSINESS_QUEUEA_NAME).withArguments(args).build();
    }

    // 声明业务队列B
    @Bean("businessQueueB")
    public Queue businessQueueB(){
        Map<String, Object> args = new HashMap<>(2);
//       x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
//       x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEB_ROUTING_KEY);
        return QueueBuilder.durable(BUSINESS_QUEUEB_NAME).withArguments(args).build();
    }

    // 声明死信队列A
    @Bean("deadLetterQueueA")
    public Queue deadLetterQueueA(){
        return new Queue(DEAD_LETTER_QUEUEA_NAME);
    }

    // 声明死信队列B
    @Bean("deadLetterQueueB")
    public Queue deadLetterQueueB(){
        return new Queue(DEAD_LETTER_QUEUEB_NAME);
    }

    // 声明业务队列A绑定关系
    @Bean
    public Binding businessBindingA(@Qualifier("businessQueueA") Queue queue,
                                    @Qualifier("businessExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

    // 声明业务队列B绑定关系
    @Bean
    public Binding businessBindingB(@Qualifier("businessQueueB") Queue queue,
                                    @Qualifier("businessExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

    // 声明死信队列A绑定关系
    @Bean
    public Binding deadLetterBindingA(@Qualifier("deadLetterQueueA") Queue queue,
                                    @Qualifier("deadLetterExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEA_ROUTING_KEY);
    }

    // 声明死信队列B绑定关系
    @Bean
    public Binding deadLetterBindingB(@Qualifier("deadLetterQueueB") Queue queue,
                                      @Qualifier("deadLetterExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEB_ROUTING_KEY);
    }
}

业务队列的消费代码

@RabbitListener(queues = BUSINESS_QUEUEA_NAME)
    public void receiveA(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("收到业务消息A：{}", msg);
        boolean ack = true;
        Exception exception = null;
        try {
            if (msg.contains("deadletter")){
                throw new RuntimeException("dead letter exception");
            }
        } catch (Exception e){
            ack = false;
            exception = e;
        }
        if (!ack){
            log.error("消息消费发生异常，error msg:{}", exception.getMessage(), exception);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } else {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    @RabbitListener(queues = BUSINESS_QUEUEB_NAME)
    public void receiveB(Message message, Channel channel) throws IOException {
        System.out.println("收到业务消息B：" + new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

配置死信队列的消费者

@Component
public class DeadLetterMessageReceiver {


    @RabbitListener(queues = DEAD_LETTER_QUEUEA_NAME)
    public void receiveA(Message message, Channel channel) throws IOException {
        System.out.println("收到死信消息A：" + new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = DEAD_LETTER_QUEUEB_NAME)
    public void receiveB(Message message, Channel channel) throws IOException {
        System.out.println("收到死信消息B：" + new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

7.3.4 死信消息的变化

那么“死信”被丢到死信队列中后，会发生什么变化呢？

如果队列配置了参数 x-dead-letter-routing-key 的话，“死信”的路由key将会被替换成该参数对应的值。如果没有设置，则保留该消息原有的路由key。

举个栗子：

如果原有消息的路由key是testA，被发送到业务Exchage中，然后被投递到业务队列QueueA中，如果该队列没有配置参数x-dead-letter-routing-key，则该消息成为死信后，将保留原有的路由keytestA，如果配置了该参数，并且值设置为testB，那么该消息成为死信后，路由key将会被替换为testB，然后被抛到死信交换机中。

另外，由于被抛到了死信交换机，所以消息的Exchange Name也会被替换为死信交换机的名称。消息的Header中，也会添加很多奇奇怪怪的字段，修改一下上面的代码，在死信队列的消费者中添加一行日志输出：

log.info("死信消息properties：{}", message.getMessageProperties());

然后重新运行一次，即可得到死信消息Header中被添加的信息：

死信消息
properties：MessageProperties [headers={x-first-death-exchange=dead.letter.demo.simple.business.exchange, x-death=[{reason=rejected, count=1, exchange=dead.letter.demo.simple.business.exchange, time=Sun Jul 14 16:48:16 CST 2019, routing-keys=[], queue=dead.letter.demo.simple.business.queuea}], x-first-death-reason=rejected, x-first-death-queue=dead.letter.demo.simple.business.queuea}, correlationId=1, replyTo=amq.rabbitmq.reply-to.g2dkABZyYWJiaXRAREVTS1RPUC1DUlZGUzBOAAAPQAAAAAAB.bLbsdR1DnuRSwiKKmtdOGw==, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=dead.letter.demo.simple.deadletter.exchange, receivedRoutingKey=dead.letter.demo.simple.deadletter.queuea.routingkey, deliveryTag=1, consumerTag=amq.ctag-NSp18SUPoCNvQcoYoS2lPg, consumerQueue=dead.letter.demo.simple.deadletter.queuea]

Header中看起来有很多信息，实际上并不多，只是值比较长而已。下面就简单说明一下Header中的值：

7.3.5 消息的生命周期

总结一下死信消息的生命周期：

业务消息被投入业务队列
消费者消费业务队列的消息，由于处理过程中发生异常，于是进行了nck或者reject操作
被nck或reject的消息由RabbitMQ投递到死信交换机中
死信交换机将消息投入相应的死信队列
死信队列的消费者消费死信消息

死信消息是RabbitMQ为我们做的一层保证，其实我们也可以不使用死信队列，而是在消息消费异常时，将消息主动投递到另一个交换机中，当你明白了这些之后，这些Exchange和Queue想怎样配合就能怎么配合。比如从死信队列拉取消息，然后发送邮件、短信、钉钉通知来通知开发人员关注。或者将消息重新投递到一个队列然后设置过期时间，来进行延时消

7.3.6 死信对列的使用场景

一般用在较为重要的业务队列中，确保未被正确消费的消息不被丢弃，一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常，处理过程中参数校验异常，或者因网络波动导致的查询异常等等，当发生异常时，当然不能每次通过日志来获取原消息，然后让运维帮忙重新投递消息（没错，以前就是这么干的= =）。通过配置死信队列，可以让未正确处理的消息暂存到另一个队列中，待后续排查清楚问题后，编写相应的处理代码来处理死信消息，这样比手工恢复数据要好太多了

8.RbbitMq的TTL

TTL（Time To Live） TTL是什么呢？TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了TTL属性或者进入了设置TTL属性的队列，那么这条消息如果在TTL设置的时间内没有被消费，则会成为“死信” 。如果同时配置了队列的TTL和消息的TTL，那么较小的那个值将会被使用。

那么，如何设置这个TTL值呢？有两种方式，第一种是在创建队列的时候设置队列的“x-message-ttl”属性，如下：

Map<String, Object> args = new HashMap<String, Object>();
// 申明队列的时候设置整个队列的消息超时时间
args.put("x-message-ttl", 6000);
channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

这样所有被投递到该队列的消息都最多不会存活超过6s。

另一种方式便是针对每条消息设置TTL，代码如下：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.expiration("6000");
AMQP.BasicProperties properties = builder.build();
channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

这两种方式是有区别的，如果设置了队列的TTL属性，那么一旦消息过期，就会被队列丢弃，而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间

另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

TTL:过期时间设置

队列统一过期
消息单独过期（推荐使用这种，更通用）

如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。

队列过期后，会将队列所有消息全部移除
消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉)

9.消息可靠性100%投递

RabbitMQ消息的可靠性投递主要两种实现：

1、通过实现消费的重试机制，通过@Retryable来实现重试，可以设置重试次数和重试频率；

2、生产端实现消息可靠性投递。

两种方法消费端都可能收到重复消息，要求消费端必须实现幂等性消费。

消息幂等性处理：通过消息唯一性标志将已消费的消息入库，mysql/redis中，每次消费根据这个唯一性标志判断是否已经消费过来该条消息，消费成功则入库，消费失败不入库。

9.1 生产者端

9.1.1 mandatory 参数

设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。当把 mandotory 参数设置为 true 时，如果交换机无法将消息进行路由时，会将该消息返回给生产者，而如果该参数设置为false，如果发现消息无法进行路由，则直接丢弃。

那么如何设置这个参数呢？在发送消息的时候，只需要在初始化方法添加一行代码即可：

rabbitTemplate.setMandatory(true);

设置 mandatory 参数后，如果消息无法被路由，消费者后台日志输出中会有 Returned message but no callback available 信息打印，则会返回给生产者，是通过回调的方式进行的，所以，生产者需要设置相应的回调函数才能接受该消息。

交换机、队列及其绑定：

public static final String BUSINESS_EXCHANGE_NAME = "rabbitmq.tx.demo.simple.business.exchange";
public static final String BUSINESS_QUEUEA_NAME = "rabbitmq.tx.demo.simple.business.queue";

// 声明业务 Exchange
@Bean("businessExchange")
public DirectExchange businessExchange(){
    return new DirectExchange(BUSINESS_EXCHANGE_NAME);
}

// 声明业务队列
@Bean("businessQueue")
public Queue businessQueue(){
    return QueueBuilder.durable(BUSINESS_QUEUEA_NAME).build();
}

// 声明业务队列绑定关系
@Bean
public Binding businessBinding(@Qualifier("businessQueue") Queue queue,
                               @Qualifier("businessExchange") DirectExchange exchange){
    return BindingBuilder.bind(queue).to(exchange).with("key");
}

生产者发送消息：

@Autowired
private RabbitTemplate rabbitTemplate;

@PostConstruct
private void init() {
    //        rabbitTemplate.setChannelTransacted(true);
    rabbitTemplate.setConfirmCallback(this);
}

public void sendCustomMsg(String exchange, String msg) {
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

    log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

    rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

    correlationData = new CorrelationData(UUID.randomUUID().toString());

    log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

    rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
}

@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
    String id = correlationData != null ? correlationData.getId() : "";
    if (b) {
        log.info("消息确认成功, id:{}", id);
    } else {
        log.error("消息未成功投递, id:{}, cause:{}", id, s);
    }
}

生产者实现一个接口 RabbitTemplate.ReturnCallback


@Slf4j
@Component
public class BusinessMsgProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    private void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    public void sendCustomMsg(String exchange, String msg) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

        correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (b) {
            log.info("消息确认成功, id:{}", id);
        } else {
            log.error("消息未成功投递, id:{}, cause:{}", id, s);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("消息被服务器退回。msg:{}, replyCode:{}. replyText:{}, exchange:{}, routingKey :{}",
                new String(message.getBody()), replyCode, replyText, exchange, routingKey);
    }
}

9.1.2 备份交换机

防止生产者投递消息路由失败，可以搞一个像死信对列的备胎，这里是搞一个备份交换机来做备胎

什么是备份交换机呢？

备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会将这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。

备份交换机 :


@Configuration
public class RabbitMQConfig {

    public static final String BUSINESS_EXCHANGE_NAME = "rabbitmq.backup.test.exchange";
    public static final String BUSINESS_QUEUE_NAME = "rabbitmq.backup.test.queue";
    public static final String BUSINESS_BACKUP_EXCHANGE_NAME = "rabbitmq.backup.test.backup-exchange";
    public static final String BUSINESS_BACKUP_QUEUE_NAME = "rabbitmq.backup.test.backup-queue";
    public static final String BUSINESS_BACKUP_WARNING_QUEUE_NAME = "rabbitmq.backup.test.backup-warning-queue";

    // 声明业务 Exchange
    @Bean("businessExchange")
    public DirectExchange businessExchange(){
        ExchangeBuilder exchangeBuilder = ExchangeBuilder.directExchange(BUSINESS_EXCHANGE_NAME)
                .durable(true)
                .withArgument("alternate-exchange", BUSINESS_BACKUP_EXCHANGE_NAME);

        return (DirectExchange)exchangeBuilder.build();
    }

    // 声明备份 Exchange
    @Bean("backupExchange")
    public FanoutExchange backupExchange(){
        ExchangeBuilder exchangeBuilder = ExchangeBuilder.fanoutExchange(BUSINESS_BACKUP_EXCHANGE_NAME)
                .durable(true);
        return (FanoutExchange)exchangeBuilder.build();
    }

    // 声明业务队列
    @Bean("businessQueue")
    public Queue businessQueue(){
        return QueueBuilder.durable(BUSINESS_QUEUE_NAME).build();
    }

    // 声明业务队列绑定关系
    @Bean
    public Binding businessBinding(@Qualifier("businessQueue") Queue queue,
                                    @Qualifier("businessExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("key");
    }

    // 声明备份队列
    @Bean("backupQueue")
    public Queue backupQueue(){
        return QueueBuilder.durable(BUSINESS_BACKUP_QUEUE_NAME).build();
    }

    // 声明报警队列
    @Bean("warningQueue")
    public Queue warningQueue(){
        return QueueBuilder.durable(BUSINESS_BACKUP_WARNING_QUEUE_NAME).build();
    }

    // 声明备份队列绑定关系
    @Bean
    public Binding backupBinding(@Qualifier("backupQueue") Queue queue,
                                   @Qualifier("backupExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

    // 声明备份报警队列绑定关系
    @Bean
    public Binding backupWarningBinding(@Qualifier("warningQueue") Queue queue,
                                 @Qualifier("backupExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

}

这里我们使用 ExchangeBuilder 来创建交换机，并为其设置备份交换机：

 .withArgument("alternate-exchange", BUSINESS_BACKUP_EXCHANGE_NAME);

为业务交换机绑定了一个队列，为备份交换机绑定了两个队列，一个用来存储不可投递消息，待之后人工处理，一个专门用来做报警用途。

接下来，分别为业务交换机和备份交换机创建消费者：

@Slf4j
@Component
public class BusinessMsgConsumer {

    @RabbitListener(queues = BUSINESS_QUEUE_NAME)
    public void receiveMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("收到业务消息：{}", msg);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    }
}

@Slf4j
@Component
public class BusinessWaringConsumer {

    @RabbitListener(queues = BUSINESS_BACKUP_WARNING_QUEUE_NAME)
    public void receiveMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.error("发现不可路由消息：{}", msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

接下来我们分别发送一条可路由消息和不可路由消息：

@Slf4j
@Component
public class BusinessMsgProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendCustomMsg(String exchange, String msg) {


        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());


        log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

        correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
    }
}

消息如下：

消息id:5c3a33c9-0764-4d1f-bf6a-a00d771dccb4, msg:1
消息id:42ac8c35-1d0a-4413-a1df-c26a85435354, msg:1
收到业务消息：1
发现不可路由消息：1

这里仅仅使用 error 日志配合日志系统进行报警，如果是敏感数据，可以使用邮件、钉钉、短信、电话等报警方式来提高时效性。

那么问题来了，mandatory 参数与备份交换机可以一起使用吗？设置 mandatory 参数会让交换机将不可路由消息退回给生产者，而备份交换机会让交换机将不可路由消息转发给它，那么如果两者同时开启，消息究竟何去何从？？

emmm，想这么多干嘛，试试不就知道了。

修改一下生产者即可：

@Slf4j
@Component
public class BusinessMsgProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    private void init() {
//        rabbitTemplate.setChannelTransacted(true);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    public void sendCustomMsg(String exchange, String msg) {


        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

        correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("消息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (b) {
            log.info("消息确认成功, id:{}", id);
        } else {
            log.error("消息未成功投递, id:{}, cause:{}", id, s);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("消息被服务器退回。msg:{}, replyCode:{}. replyText:{}, exchange:{}, routingKey :{}",
                new String(message.getBody()), replyCode, replyText, exchange, routingKey);
    }
}

再来测试一下：

消息id:0a3eca1e-d937-418c-a7ce-bfb8ce25fdd4, msg:1
消息id:d8c9e010-e120-46da-a42e-1ba21026ff06, msg:1
消息确认成功, id:0a3eca1e-d937-418c-a7ce-bfb8ce25fdd4
消息确认成功, id:d8c9e010-e120-46da-a42e-1ba21026ff06
发现不可路由消息：1
收到业务消息：1

可以看到，两条消息都可以收到确认成功回调，但是不可路由消息不会被回退给生产者，而是直接转发给备份交换机。可见备份交换机的处理优先级更高。

9.1.3 confirm 确认模式

yml配置

spring: #rabbitmq 连接配置
  rabbitmq: 
    publisher-confirm-type: correlated # 开启confirm确认模式

实现confirm方法

实现ConfirmCallback接口中的confirm方法，消息只要被 rabbitmq broker接收到就会触 ConfirmCallback 回调，ack为true表示消息发送成功，ack为false表示消息发送失败

import org.springframework.amqp.rabbit.core.RabbitTemplate; 
import org.springframework.stereotype.Component; 
/*** 实现ConfirmCallback接口 */
@Component 
public class ConfirmCallbackService implements RabbitTemplate.ConfirmCallback { 
    /*** @param correlationData 相关配置信息 
    * @param ack exchange交换机 是否成功收到了消息。true 成功，false代表失败 
    * @param cause 失败原因 
    */
    @Override 
    public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
        if (ack) { 
        //接收成功 
            System.out.println("成功发送到交换机<===>"); 
        } else { 
            //接收失败 
            System.out.println("失败原因:===>" + cause);
            //TODO 做一些处理:消息再次发送等等 
        } 
    } 
}

9.1.4 return 退回模式

yml配置：

spring: # rabbitmq 连接配置 
   rabbitmq: publisher-returns: true # 开启退回模式

设置投递失败的模式

如果消息没有路由到Queue，则丢弃消息（默认）

如果消息没有路由到Queue，返回给消息发送方ReturnCallBack（开启后）
```
rabbitTemplate.setMandatory(true);
```

实现returnedMessage方法

package com.rabbitmq.config; 
import org.springframework.amqp.core.Message; 
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component; 
@Component 
public class ReturnCallbackService implements RabbitTemplate.ReturnCallback { 
    /*** @param message 消息对象 
    * @param replyCode 错误码 
    * @param replyText 错误信息 
    * @param exchange 交换机 
    * @param routingKey 路由键 
    */
    @Override 
    public void returnedMessage(Message message, int replyCode, String replyText,      String exchange, String routingKey) { 
        System.out.println("消息对象===>:" + message); 
        System.out.println("错误码===>:" + replyCode); 
        System.out.println("错误信息===>:" + replyText); 
        System.out.println("消息使用的交换器===>:" + exchange); 
        System.out.println("消息使用的路由key===>:" + routingKey);
        //TODO ===>做业务处理 
    } 
 }

9.1.5 RabbitMQ的事务机制

RabbitMQ是支持AMQP事务机制的，在生产者确认机制之前，事务是确保消息被成功投递的唯一方法。

在SpringBoot项目中，使用RabbitMQ事务其实很简单，只需要声明一个事务管理的Bean，并将RabbitTemplate的事务设置为true即可。

RabbitMQConfig如下：


@Configuration
public class RabbitMQConfig {

    public static final String BUSINESS_EXCHANGE_NAME = "rabbitmq.tx.demo.simple.business.exchange";
    public static final String BUSINESS_QUEUEA_NAME = "rabbitmq.tx.demo.simple.business.queue";

    // 声明业务Exchange
    @Bean("businessExchange")
    public FanoutExchange businessExchange(){
        return new FanoutExchange(BUSINESS_EXCHANGE_NAME);
    }

    // 声明业务队列
    @Bean("businessQueue")
    public Queue businessQueue(){
        return QueueBuilder.durable(BUSINESS_QUEUEA_NAME).build();
    }

    // 声明业务队列绑定关系
    @Bean
    public Binding businessBinding(@Qualifier("businessQueue") Queue queue,
                                    @Qualifier("businessExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }


    /**
     * 配置启用rabbitmq事务
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }
}

生产者：

@Slf4j
@Component
public class BusinessMsgProducer{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    private void init() {
        rabbitTemplate.setChannelTransacted(true);
    }

    @Transactional
    public void sendMsg(String msg) {
        rabbitTemplate.convertAndSend(BUSINESS_EXCHANGE_NAME, "key", msg);
        log.info("msg:{}", msg);
        if (msg != null && msg.contains("exception"))
            throw new RuntimeException("surprise!");
        log.info("消息已发送 {}" ,msg);
    }
}

消费者：

@Slf4j
@Component
public class BusinessMsgConsumer {

    @RabbitListener(queues = BUSINESS_QUEUEA_NAME)
    public void receiveMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("收到业务消息：{}", msg);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    }
}

这里有两个注意的地方：

在初始化方法里，通过使用rabbitTemplate.setChannelTransacted(true); 来开启事务。
在发送消息的方法上加上 @Transactional 注解，这样在该方法中发生异常时，消息将不会发送。

9.2 消费者端

消息确认机制ack

ack指Acknowledge确认。表示消费端收到消息后的确认方式

消费端消息的确认分为：自动确认（默认）、手动确认、不确认

AcknowledgeMode.NONE：不确认

AcknowledgeMode.AUTO：自动确认

AcknowledgeMode.MANUAL：手动确认

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

1、yml配置

spring: rabbitmq: 
  listener: simple:
    acknowledge-mode: manual # 手动确认

2、费者手动ack

@Component 
@RabbitListener(queues = "confirm_test_queue") 
public class ReceiverMessage { 
    @RabbitHandler 
    public void processHandler(String msg, Channel channel, Message message) throws IOException { 
        long deliveryTag = message.getMessageProperties().getDeliveryTag(); 
        try {System.out.println("消息内容===>" + new String(message.getBody())); 
             //TODO 具体业务逻辑 
             //手动签收[参数1:消息投递序号,参数2:批量签收]
             channel.basicAck(deliveryTag, true); } 
        catch (Exception e) { 
            //拒绝签收[参数1:消息投递序号,参数2:批量拒绝,参数3:是否重新加入队列]
            channel.basicNack(deliveryTag, true, true);
        } 
    } 
}

channel.basicNack 方法与 channel.basicReject 方法区别在于basicNack可以批量拒绝多条消息，而 basicReject一次只能拒绝一条消息。

注意：如果消费者开启手动ack的话，消费消息中有异常抛出，代码中没有加try/catch,该条消息会被brock重复投递给消费者，消费者消费一直重复消费，然后出现死循环，对业务产生影响，后台会疯狂的打印该条消息的消费日志和报错消息

解决办法：

1.消费者方法代码中加入try/catch,然后消费成功或异常catch中手动ack

2.改重试机制策略，默认是一直重试

yml配置

spring:
  rabbitmq:
  ####连接地址
    host: 127.0.0.1
   ####端口号   
    port: 5672
   ####账号 
    username: guest
   ####密码  
    password: guest
   ### 地址
    virtual-host: /admin_host
    listener:
      simple:
        retry:
        ####开启消费者（程序出现异常的情况下会）进行重试
          enabled: true
         ####最大重试次数
          max-attempts: 5
        ####重试间隔次数
          initial-interval: 3000
        ####开启手动ack  
        acknowledge-mode: manual

10.持久化机制和内存磁盘控制

10.1 持久化

交换机定义时设置持久化参数为true

package org.springframework.amqp.core;

import java.util.Map;


/**
 * Common properties that describe all exchange types.
 * <p>Subclasses of this class are typically used with administrative operations that declare an exchange.
 *
 * @author Mark Pollack
 * @author Gary Russell
 * @author Artem Bilan
 *
 * @see AmqpAdmin
 */
public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {

    private final String name;

    private final boolean durable;

    private final boolean autoDelete;

    private boolean delayed;

    private boolean internal;

    /**
     * Construct a new durable, non-auto-delete Exchange with the provided name.
     * @param name the name of the exchange.
     */
    public AbstractExchange(String name) {
        this(name, true, false);
}

    /**
     * Construct a new Exchange, given a name, durability flag, auto-delete flag.
     * @param name the name of the exchange.
     * @param durable true if we are declaring a durable exchange (the exchange will
     * survive a server restart)
     * @param autoDelete true if the server should delete the exchange when it is no
     * longer in use
     */
    public AbstractExchange(String name, boolean durable, boolean autoDelete) {
        this(name, durable, autoDelete, null);
}

    /**
     * Construct a new Exchange, given a name, durability flag, and auto-delete flag, and
     * arguments.
     * @param name the name of the exchange.
     * @param durable true if we are declaring a durable exchange (the exchange will
     * survive a server restart)
     * @param autoDelete true if the server should delete the exchange when it is no
     * longer in use
     * @param arguments the arguments used to declare the exchange
     */
    public AbstractExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        super(arguments);
        this.name = name;
        this.durable = durable;
        this.autoDelete = autoDelete;
}
    ...............
        .............
}

上面是所有换机的公共抽象类：

在定义交换机时指定durable参数为true即即开启持久化

10.2 内存告警

默认情况下 set_vm_memory_high_watermark 的值为 0.4，即内存阈值（临界值）为 0.4，表示当RabbitMQ 使用的内存超过 40%时，就会产生内存告警并阻塞所有生产者的连接。一旦告警被解除(有消息被消费或者从内存转储到磁盘等情况的发生)，一切都会恢复正常。

在出现内存告警后，所有的客户端连接都会被阻塞。阻塞分为 blocking 和 blocked 两种。

blocking：表示没有发送消息的链接。

blocked：表示试图发送消息的链接

如果出现了内存告警，并且机器还有可用内存，可以通过命令调整内存阈值，解除告警。

rabbitmqctl set_vm_memory_high_watermark 1 1

或者

rabbitmqctl set_vm_memory_high_watermark absolute 1GB

但这种方式只是临时调整，RabbitMQ 服务重启后，会还原。如果需要永久调整，可以修改配置文件。但修改配置文件需要重启**RabbitMQ** 服务才能生效。

修改配置文件：vim /etc/rabbitmq/rabbitmq.conf

vm_memory_high_watermark.relative = 0.4

或者

vm_memory_high_watermark.absolute = 1GB

10.3 内存告警模拟

调整内存阈值，模拟出告警，在RabbitMQ 服务器上修改。注意：修改之前，先在管理页面看一下当前使用了多少，调成比当前值小

rabbitmqctl set_vm_memory_high_watermark absolute 50MB 1

刷新管理页面（可能需要刷新多次），在 Overview -> Nodes 中可以看到Memory变成了红色，表示此节点内存告警了

启动 Producer 和 Consumer

查看管理界面的 Connections 页面，可以看到生产者和消费者的链接都处于 blocking 状态。

在 Producer 的控制台按回车健，再观察管理界面的 Connections 页面，会发现生产者的状态成了 blocked 。

此时虽然在 Producer 控制台看到了发送两条消息的信息，但 Consumer 并没有收到任何消息。并且在管理界面的 Queues 页面也看到不到队列的消息数量有变化。

解除内存告警后，会发现 Consumer 收到了 Producer 发送的两条消息。

10.4 内存换页

在Broker节点的使用内存即将达到内存阈值之前，它会尝试将队列中的消息存储到磁盘以释放内存空间，这个动作叫内存换页。

持久化和非持久化的消息都会被转储到磁盘中，其中持久化的消息本身就在磁盘中有一份副本，此时会将持久化的消息从内存中清除掉。

默认情况下，在内存到达内存阈值的 50%时会进行换页动作。也就是说，在默认的内存阈值为 0.4 的情况下，当内存超过 0.4 x 0 .5=0.2 时会进行换页动作。

通过修改配置文件，调整内存换页分页阈值（不能通过命令调整）。

# 此值大于1时，相当于禁用了换页功能。
vm_memory_high_watermark_paging_ratio = 0.75

10.5 磁盘告警

当磁盘剩余空间低于磁盘的阈值时，RabbitMQ 同样会阻塞生产者，这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务崩溃

默认情况下，磁盘阈值为50MB，表示当磁盘剩余空间低于50MB 时会阻塞生产者并停止内存中消息的换页动作

这个阈值的设置可以减小，但不能完全消除因磁盘耗尽而导致崩渍的可能性。比如在两次磁盘空间检测期间内，磁盘空间从大于50MB被耗尽到0MB

通过命令可以调整磁盘阈值，临时生效，重启恢复

# disk_limit 为固定大小，单位为MB、GB 
rabbitmqctl set_disk_free_limit <disk_limit>

或者：

# fraction 为相对比值，建议的取值为1.0~2.0之间 
rabbitmqctl set_disk_free_limit mem_relative <fraction>

10.6 磁盘告警模拟

在服务器通过命令，临时调整磁盘阈值（需要设置一个绝对大与当前磁盘空间的数值）
```
rabbitmqctl set_disk_free_limit 102400GB
```
刷新管理页面（可能需要刷新多次），在 Overview -> Nodes 中可以看到Disk space变成了红色，表示此节点磁盘告警了
后续步骤同模拟内存告警。

11.rabbitmqctl命令

# 启动服务 
rabbitmq-server 
# 停止服务 
rabbitmqctl stop 
# vhost 增删查 
rabbitmqctl add_vhost 
rabbitmqctl delete_vhost 
rabbitmqctl list_vhosts 
# 查询交换机 
rabbitmqctl list_exchanges 
# 查询队列 
rabbitmqctl list_queues 
# 查看消费者信息 
rabbitmqctl list_consumers 
# user 增删查 
rabbitmqctl add_user 
rabbitmqctl delete_user 
rabbitmqctl list_users

net start RabbitMQ  启动
net stop RabbitMQ  停止
rabbitmqctl status  查看状态

健康检查：rabbitmqctl status

启动监控管理器：rabbitmq-plugins enable rabbitmq_management

关闭监控：rabbitmq-plugins disable rabbitmq_management

停止服务：rabbitmq-service stop

启动服务：rabbitmq-service start

重启命令：net stop RabbitMQ && net start

帮助命令：rabbitmqctl help

rabbitmqctl list_queues查看所有队列

rabbitmqctl reset清除所有队列

rabbitmqctl list_exchanges查看所有交换器

rabbitmqctl add_user username password添加用户

rabbitmqctl set_user_tags username administrator分配角色

rabbitmqctl list_bindings 查看交换器和队列的绑定关系

12.REST API

下面是一些常用的 API：

# 概括信息 
curl -i -u guest:guest http://localhost:15672/api/overview
# vhost 列表 
curl -i -u guest:guest http://localhost:15672/api/vhosts 
# channel 列表 
curl -i -u guest:guest http://localhost:15672/api/channels 
# 节点信息 
curl -i -u guest:guest http://localhost:15672/api/nodes 
# 交换机信息 
curl -i -u guest:guest http://localhost:15672/api/exchanges 
# 队列信息 
curl -i -u guest:guest http://localhost:15672/api/queues

13.总结

到此，关于RabbitMq的相关的东西就已经全部已经串联起来了，方便新手学习和避坑指南，请一键三连加关注哦

ITPub博客

RabbitMq实现延迟队列的正确姿势

1.1 采用docker方式安装

1.2 windows环境下安装

1.3 Linux环境下安装（CentOs7.x）

1.3.1 安装依赖环境

1.3.2 安装RabbitMq

1.3.3 启动和关闭

1.3.4 开启web管理插件

1.3.5 防火墙端口配置

1.3.6 RabbitMQ基本配置

1.3.7 RabbitMQ管理界面

1.3.8 RabbitMQ角色分类

1.4 多机多节点集群方式部署

1.4.1 环境准备

1.4.2 修改配置文件

1.4.3 防火墙端口配置

1.4.4 启动RabbitMq

1.5 单机多节点部署

1.6 镜像队列模式集群部署

2.延迟插件安装

2.1 插件官网下载地址

2.2 docker容器内安装延迟插件

3.RabbitMq的概念模型

3.1 什么是消息队列？

3.2 为什么要用消息队列？

3.3 RabbitMq的特点？

3.3.1 特点一总结

3.3.1.1 可靠性（Reliability）

3.3.1.2 灵活的路由（Flexible Routing）

3.3.1.3 消息集群（Clustering）

3.3.1.4 高可用（Highly Available Queues）

3.3.1.5 多种协议（Multi-protocol）

3.3.1.6 多语言客户端（Many Clients）

3.3.1.7 管理界面（Management UI）

3.3.1.8 跟踪机制（Tracing）

3.3.1.9 插件机制（Plugin System）

3.3.2 特点二总结

3.3.2.1 RabbitMQ底层使用Erlang语言编写，传递效率高，延迟低

3.3.2.2 开源、性能优秀、稳定性较高

3.3.2.3 与SpringAMQP完美的整合、API丰富

3.3.2.4 集群模式丰富、表达式配置、HA模式、镜像队列模式

3.3.2.5 保证数据不丢失的情况下，做到高可用

3.3.2.6 AMQP全称：Advanced Message Queuing Protocol

3.3.2.7 AMQP翻译:高级消息队列协议

3.4 消息模型

3.5 RabbitMQ 内部结构

3.5.1 Message

3.5.2 Publisher

3.5.3 Exchange

3.5.4 Binding

3.5.5 Queue

3.5.6 Connection

3.5.7 Channel

3.5.8 Consumer

3.5.9 Virtual Host

3.5.10 Broker

3.6 消息路由

3.6.1 AMQP 的消息路由过程

3.7 交换机类型

3.7.1 direct

3.7.2 fanout

3.7.3 topic

3.8 RabbitMQ提供了7 种模式

4.什么是延迟队列？以及它的应用场景？

5.项目集成

5.1 pom依赖

5.2 yml配置

6.错误姿势

7.正确姿势

7.1 延迟插件实现

7.2 死信队列实现延迟队列

7.2.1 什么是死信队列？

7.2.2 消息成为死信的三种情况

7.2.3 队列绑定死信交换机

7.2.4 TTL+死信队列 组合实现延迟队列

7.3 死信对列的使用

7.3.1 消息成为死信的三种情况

7.3.2 如何配置死信队列呢？

7.3.3 死信对列没有延迟的使用

7.2.4 TTL+死信队列组合实现延迟队列