文章目录

RabbitMQ学习笔记
- 前言
- 1、RabbitMQ概述
- - 1.1 MQ概述
  - 1.2 RabbitMQ概述
- 2、RabbitMQ的安装
- 3、RabbitMQ初体验
- 4、工作队列
- - 4.1 消息应答
  - - 4.1.1 消息应答简介
    - 4.1.2 消息手动应答
  - 4.2 队列和消息的持久化
  - - 4.2.1 队列持久化
    - 4.2.2 消息持久化
  - 4.3 消息分发
  - - 4.3.1 不公平分发
    - 4.3.2 预取值
  - 4.4 发布和确认
  - - 4.4.1 发布确认概述
    - 4.4.2 发布确认案例
- 5、交换机
- - 5.1 交换机概述
  - 5.2 Fanout 交换机
  - 5.3 Direct 交换机
  - 5.4 Topic 交换机
- 6、死信队列
- - 6.1 死信队列概述
  - 6.2 死信队列案例
- 7、延迟队列
- - 7.1 延迟队列概述
  - 7.2 基于死信队列实现延迟队列
  - 7.3 延迟队列优化
  - 7.4 基于插件实现延迟队列
- 8、发布确认高级
- - 8.1 发布确认高级介绍
  - 8.2 发布确认高级案例
- 9、备份交换机
- 10、其它队列
- - 10.1 幂等性
  - 10.2 优先级队列
  - 10.3 惰性队列
- 11、MQ集群
- - 11.1 普通模式
  - 11.2 镜像模式
  - 11.3 实现集群的高可用
  - 11.4 实现消息的可靠传递和数据的同步
  - - 11.4.1 Federation
    - 11.4.2 Shovel
总结

RabbitMQ学习笔记

[RabbitMQ]AMQP 0-9-1：模型

前言

本文是笔者学习RabbitMQ的笔记，如果对你有所帮助，欢迎三连（点赞👍+收藏⭐+关注💖）
在现代的分布式系统中，消息传递变得愈发重要，而RabbitMQ作为一种高性能、可靠且灵活的消息队列解决方案备受青睐。作为开源软件，RabbitMQ提供了一个可靠的、可扩展的平台，用于在应用程序之间传递消息，无论是在单个应用程序内部还是跨多个应用程序之间。

本文将介绍RabbitMQ以及与其相关的学习内容。首先，我们将学习RabbitMQ中的相关核心概念，接下来，我们将介绍学习如何简单的使用消息队列，以及如何操作RabbitMQ客户端，最后我们将学习如何搭建MQ集群。此外，我们还将探讨RabbitMQ的关键特性，如消息持久化、发布/订阅模式、消息路由和负载均衡等。

无论你是一个新手想要了解消息队列的基础知识，还是一个有经验的开发者希望在分布式系统中应用消息传递，本文都将为你提供有价值的信息。让我们走进RabbitMQ的世界，探索它的强大功能和无限潜力，为你带来更强大的应用程序开发体验。

1、RabbitMQ概述

1.1 MQ概述

什么是MQ？
- MQ（message queue，消息队列），从字面意思上看，本质是个队列，特点是FIFO （先入先出），只不过队列中存放的内容是 message 而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ 是一种非常常见的上下游「逻辑解耦 + 物理解耦」的消息通信服务。使用了 MQ 之后，消息发送上游只需要依赖 MQ，不用依赖其他服务。
  
  综上所诉，MQ（Message Queue）是一种用于消息传输的软件架构，通常用于在分布式系统中传输数据或消息，解决系统间的异步通信问题。
- 个人理解：MQ是一个存放消息的容器，这个容器符合FIFO的特点，即生产者生产消息，将消息放入MQ中，然后消费者从MQ中取消息，存和取满足先进先出原则
什么是消息？
- “消息”一词在电脑运算上有两种主要意义：一种是由电脑系统本身发送而属于其人类用户之间的消息；另一种是为特定目的在不同计算机程序之间或一支程序的不同组件之间互相发送的消息。——维基百科
- 个人理解：在软件开发中，消息就是指蕴含了某种意义的数据，这个数据能让程序直到接下来要干什么，而消息的形式也是多种多样的，比如：对象、变量、函数、常量……都可以当作一个消息
消息的分类
- 同步消息：发送的消息需要等待接收方作出响应，然后才能接着发送消息
- 异步消息：发送的消息不需要接收方作出响应，可以继续发送消息
Java处理消息的三种异步消息传递技术
- JMS：即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。
  
  JMS规范中规范了消息有两种模型。分别是点对点模型和发布订阅模型
  - 点对点模型：peer-2-peer，生产者会将消息发送到一个保存消息的容器中，通常使用队列模型，使用队列保存消息。一个队列的消息只能被一个消费者消费，或未被及时消费导致超时。这种模型下，生产者和消费者是一对一绑定的
  - 发布订阅模型：publish-subscribe，生产者将消息发送到一个保存消息的容器中，也是使用队列模型来保存。但是消息可以被多个消费者消费，生产者和消费者完全独立，相互不需要感知对方的存在
  JMS中消息共有6种：：TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage、Message （只有消息头和属性）。JMS实现有：ActiveMQ、Redis、HornetMQ、RabbitMQ、RocketMQ(没有完全遵守JMS规范)
  
  备注：JMS只规范了Java语言的消息实现。JMS规范了消息开发的API
- AMQP：即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。AMQP规范了网络交换的数据格式，兼容JMS。
  
  备注：弥补了JMS的缺点，能够规范各种语言的消息实现，具有跨平台性。AMQP规范了消息的传递格式
  
  AMQP相较于JMS灵活性更高，有更多的消息模型，：
  - direct exchange：直接交换将消息中的 Routing key 与该 Exchange 关联的所有 Binding 中的 Routing key 进行比较，如果相等，则发送到该 Binding 对应的 Queue 中
  - fanout exchange：扇出交换是一种将收到的消息路由到绑定到它的所有队列的交换。当生成者将消息发送到扇出交换时，它会复制消息并路由到绑定到它的所有队列。它只是忽略路由密钥或生产者提供的任何模式匹配。当需要将同一消息存储在一个或多个队列中时，这种类型的交换很有用
  - topic exchange：主题交换和直接交换类似，都是通过routing key和binding key进行匹配，不同的是topic exchange可以为routing key设置多重标准
  - headers exchange：标头交换用于在多个上路由更容易表示为消息的属性标头而不是路由密钥。标头交换忽略路由密钥属性。相反，用于路由取自标头属性。一条消息是如果标头的值等于绑定时指定的值
  - system exchange：系统交换，Publisher向System Exchange发送 routingKey=S 的消息。System Exchange会将该消息转发给名为 S 的系统服务
  AMQP统一了消息格式，消息种类只有一种：字节数组（byte[]）
  
  AMQP的实现：RabbitMQ、StormMQ、RocketMQ
- MQTT：MQTT（Message Queueing Telemetry Transport）是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛，主要应用于物联网……
MQ的作用有哪些？
- 流量削峰。流量削峰是指在请求高峰阶段，将客户端发送的请求添加到消息队列中，然后存储在消息队列中的请求会慢慢发送给数据库，这样能够防止请求一下全部打到数据库上，导致数据库崩溃；
  
  PS：这个类似于缓存，主要目的就是为了降低数据库的压力，防止数据库一下接受到大量请求导致崩溃
- 应用解耦。应用解耦是指当一个子系统或者子模块出现异常时，并不会影响到整个系统的正常运行，这是因为当某一个子系统出现异常导致某一个请求没有被完成，系统会先将这个出现异常的请求存储到消息队列中，等异常处理完成后，系统再去处理这个因异常而未完成的请求。整个过程用户是完全未感知的，不仅保障了系统的高可用性，还提高了用户的体验
- 异步处理。异步处理是指发送完请求不需要等待响应结果即可进行下一步操作。以前一般有两种方式，A 过一段时间去调用 B 的查询 api 查询；或者 A 提供一个 callback api， B 执行完之后调用 api 通知 A 服务，这两种方式都不是很优雅。使用消息总线，可以很方便解决这个问题， A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息。
  
  如图所示。一个客户端请求发送进来，系统A会调用系统B、C、D三个子系统，同步请求的话，响应时间就是系统A、B、C、D的总和，也就是800ms。如果使用MQ，系统A发送数据到MQ，然后就可以返回响应给客户端，不需要再等待系统B、C、D的响应，可以大大地提高性能。对于一些非必要的业务，比如发送短信，发送邮件等等，就可以采用MQ。
MQ的分类：
- 按照实现方式：
  - 基于内存的消息队列：将消息存储在内存中，消息的传输速度快，适用于实时性较高的场景，如日志采集、监控告警等。
  - 基于磁盘的消息队列：将消息存储在磁盘中，可以实现消息的持久化存储，适用于数据量较大、对消息可靠性要求较高的场景，如订单处理、支付通知等。
- 按照消息模式：
  - 点对点模式：一条消息只能被一个消费者消费。
  - 发布/订阅模式：一条消息可以被多个消费者消费，适用于广播、通知等场景。
  - 路由模式：根据消息的路由键将消息发送到指定的队列中，支持多种路由规则，如直接匹配、通配符匹配、正则表达式匹配等。
- 按照消息协议：
  - AMQP：高级消息队列协议（Advanced Message Queuing Protocol），是一种跨平台的消息中间件协议。
  - MQTT：轻量级消息传输协议（Message Queuing Telemetry Transport），是一种适用于物联网设备的协议。
  - STOMP：简单文本协议（Simple Text Oriented Messaging Protocol），是一种基于文本的协议，易于实现和调试。
- 按照使用场景：
  - 异步消息处理：将消息异步地发送到消息队列中，降低系统之间的耦合性，提高系统的可扩展性和可维护性。
  - 分布式系统集成：将不同系统之间的数据进行交换和同步，实现分布式系统之间的解耦。
  - 数据缓存：将热点数据缓存在消息队列中，提高系统的响应速度和吞吐量。
  - 流式数据处理：将实时数据以流的形式发送到消息队列中，进行实时处理和分析
市面上常见的MQ产品：
- ActiveMQ
  
  ActiveMQ全称 Apache ActiveMQ ，是Apache软件基金会所研发的开放源代码消息中间件；由于ActiveMQ是一个纯 Java 程序，因此只需要操作系统支持 Java 虚拟机，ActiveMQ便可执行
  - 优点：单机吞吐量万级，时效性 ms 级，可用性高，基于主从架构实现高可用性，消息可靠性较低的概率丢失数据
  - 缺点：官方社区现在对 ActiveMQ 5.x 维护越来越少，高吞吐量场景较少使用
- Kafka
  
  大数据的杀手锏，谈到大数据领域内的消息传输，则绕不开 Kafka，这款为大数据而生的消息中间件，以其百万级 TPS 的吞吐量名声大噪，迅速成为大数据领域的宠儿，在数据采集、传输、存储的过程中发挥着举足轻重的作用。目前已经被 LinkedIn，Uber，Twitter，Netflix 等大公司所采纳。
  - 优点：性能卓越，单机写入 TPS 约在百万条/秒，最大的优点，就是吞吐量高。时效性 ms 级可用性非常高，kafka 是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用,消费者采用 Pull 方式获取消息，消息有序，通过控制能够保证所有消息被消费且仅被消费一次；有优秀的第三方Kafka Web 管理界面 Kafka-Manager；在日志领域比较成熟，被多家公司和多个开源项目使用
  - 缺点：Kafka 单机超过 64 个队列/分区，Load 会发生明显的飙高现象，队列越多，load 越高，发送消息响应时间变长，使用短轮询方式，实时性取决于轮询间隔时间，消费失败不支持重试；支持消息顺序，但是一台代理宕机后，就会产生消息乱序，社区更新较慢
- RocketMQ
  
  RocketMQ 出自阿里巴巴的开源产品，用 Java 语言实现，在设计时参考了 Kafka，并做出了自己的一些改进。被阿里巴巴广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog 分发等场景。
  - 优点：单机吞吐量十万级，可用性非常高，分布式架构,消息可以做到 0 丢失,MQ 功能较为完善，还是分布式的，扩展性好,支持 10 亿级别的消息堆积，不会因为堆积导致性能下降，源码是 java 我们可以自己阅读源码，定制自己公司的 MQ
  - 缺点：支持的客户端语言不多，目前是 java 及 c++，其中 c++ 不成熟；社区活跃度一般,没有在 MQ 核心中去实现 JMS 等接口，有些系统要迁移需要修改大量代码
- RabbitMQ
  
  2007 年发布，是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一
  - 优点：由于 Erlang 语言的高并发特性，性能较好；吞吐量到万级，MQ 功能比较完备,健壮、稳定、易用、跨平台、支持多种语言如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX 文档齐全；开源提供的管理界面非常棒，用起来很好用,社区活跃度高；更新频率相当高
  - 缺点：商业版需要收费,学习成本较高
- ZeroMQ：高性能、异步、轻量级消息库，使用C++编写，支持多种通信模式和传输协议，包括TCP、in-process等。
- Redis：内存数据库，支持发布/订阅模式和列表等数据结构，可以用作消息队列。
- Pulsar：Apache开源的分布式消息和流处理平台，使用Java编写，支持多种语言客户端和协议，包括MQTT、Kafka等。
- NSQ：分布式实时消息平台，使用Go语言编写，支持高吞吐量、水平扩展、低延迟等特性。
综上所诉：目前比比较常用的消息中间件是Kafka、RocketMQ、RabbitMQ这三款，Kafaka适合高吞吐量的场景，比如大数据领域；RocketMQ适合高可用的场景，比如金融领域；RabbitMQ适合高性能，但数据量不是特别高的场景，一般是中小型公司的首选

1.2 RabbitMQ概述

什么是RabbitMQ？
- RabbitMQ是使用Erlang语言编写的，实现了AMQP（Advanced Message Queuing Protocol，高级消息队列协议）的一款开源的消息队列服务软件（也称面向消息的中间件）
- 个人理解：MQ只是一个抽象的概念，而RabbitMQ则是实现这个概念的一个具体的产物
官方文档：RabbitMQ
RabbitMQ的作用是什么？

主要作用是存储和转发消息
RabbitMQ的特点
- 可靠性。支持持久化，传输确认，发布确认等保证了MQ的可靠性。
- 灵活的分发消息策略。这应该是RabbitMQ的一大特点。在消息进入MQ前由Exchange(交换机)进行路由消息。分发消息策略有：简单模式、工作队列模式、发布订阅模式、路由模式、通配符模式。
- 支持集群。多台RabbitMQ服务器可以组成一个集群，形成一个逻辑Broker。
- 多种协议。RabbitMQ支持多种消息队列协议，比如 STOMP、MQTT 等等。
- 支持多种语言客户端。RabbitMQ几乎支持所有常用编程语言，包括 Java、.NET、Ruby 等等。
- 可视化管理界面。RabbitMQ提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker。
- 插件机制。RabbitMQ提供了许多插件，可以通过插件进行扩展，也可以编写自己的插件。
RabbitMQ四大核心概念
- 生产者：产生数据发送消息的程序
- 交换机：是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
- 队列：是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式
- 消费者：消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。
RabbitMQ常见名词
- Broker：代理。接收和分发消息的应用，RabbitMQ Server 就是 Message Broker
- Virtual host：虚拟主机。出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个 vhost，每个用户在自己的 vhost 创建 exchange／queue 等
- Connection：连接。publisher／consumer 和 broker 之间的 TCP 连接
- Channel：信道。如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection 的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个 thread 创建单独的 channel 进行通讯，AMQP method 包含了 channel id 帮助客户端和 message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销
- Exchange：交换机。message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到 queue 中去。常用的类型有：direct (point-to-point)，topic (publish-subscribe) and fanout (multicast)
- Queue：队列。消息最终被送到这里等待 consumer 取走
- Binding：绑定。exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key，Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据
知识拓展：

我们完全可以直接使用Connection就能完成信道的工作，为什么还要引入信道呢？

试想这样一个场景，一个应用有多个线程需要从rabbitmq中消费，或是生产消息，那么必然会建立很多个connection,也就是多个tcp连接，对操作系统而言，建立和销毁tcp连接是很昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现，rabbitmq采用类似nio的做法，连接tcp连接复用，不仅可以减少性能开销，同时也便于管理

PS：要学会多问几个为什么
RabbitMQ的6大模式
- 简单模式（Simplest Mode）：也称为基本模式（Basic Mode），是最简单的模式。它只包含一个生产者、一个消费者和一个队列。生产者将消息发送到队列，消费者从队列中接收消息。
- 工作队列模式（Work Queues Mode）：也称为任务分发模式（Task Distribution Mode），它包含多个消费者和一个共享队列。生产者将消息发送到共享队列，多个消费者从队列中接收消息并进行处理。工作队列模式可以用于在分布式系统中进行任务分发和负载均衡。
  
  注意：一个队列的消息只能被消费者消费一次
- 发布/订阅模式（Publish/Subscribe Mode）：也称为广播模式（Broadcasting Mode），它包含一个生产者、多个消费者和一个交换机（Exchange）。生产者将消息发送到交换机，交换机将消息广播给所有已经绑定（Bind）到该交换机上的队列。发布/订阅模式可以用于实现广播消息和通知机制。
  
  交换机可以将同一条消息路由给不同的队列，然后消费者可以重复消费这条消息
- 路由模式（Routing Mode）：它包含一个生产者、多个消费者、一个交换机和多个队列。生产者将消息发送到交换机，交换机根据消息的路由键（Routing Key）将消息路由到匹配的队列。路由模式可以用于实现消息的有选择性地传输和过滤。
- 主题模式（Topic Mode）：也称为通配符模式（Wildcard Mode），它包含一个生产者、多个消费者、一个交换机和多个队列。生产者将消息发送到交换机，交换机根据消息的主题（Topic）将消息路由到匹配的队列。主题模式可以用于实现消息的复杂路由和匹配。
- RPC模式（Remote Procedure Call Mode）：它是一种高级模式，可以用于实现远程过程调用。它包含一个客户端、一个服务器和一个队列。客户端将请求消息发送到队列，服务器从队列中接收消息并进行处理，然后将响应消息发送回客户端。RPC模式可以用于实现分布式系统中的服务调用。
RabbitMQ的发展史以及名字的由来
- 开发阶段（2006年-2007年）：RabbitMQ最初由LShift公司（现为Nokia公司的一部分）的开发团队开发，最初称为Rabbit，用于在金融交易领域处理消息。
- 开源阶段（2007年-2010年）：RabbitMQ于2007年成为开源软件，并在GitHub上发布。该项目成为了Erlang Solutions的一部分，引起了开源社区的广泛关注。
- 成长阶段（2010年-2013年）：RabbitMQ不断改进，增加了新的特性和性能优化，如持久化、流控制、集群等。同时，Rabbit Technologies公司成立，专注于RabbitMQ的开发和支持。
- 开放标准阶段（2013年-至今）：RabbitMQ成为了AMQP 0-9-1规范的一部分，并被广泛使用。RabbitMQ还增加了对STOMP、MQTT、HTTP等协议的支持，并扩展了集群、安全、监控等方面的功能。
关于名字的由来，据说最初的开发团队考虑了多种动物的名字作为产品名称，如Beaver（海狸）和Squirrel（松鼠），最终选择了Rabbit（兔子），意为快速和敏捷。同时，该名称也与另一个开源项目Apache ActiveMQ类似，有助于用户记忆和辨别。

2、RabbitMQ的安装

这里演示在Linux中安装RabbitMQ。CentOS7、Erlang23.3.4、RabbitMQ3.8.8

Step1：下载Erlang 23.3.4

RabbitMQ是用Erlang语言开发的，所以RabbitMQ需要在Erlang环境中才能运行

注意：
1. Erlang版本要和Linux的版本对应，CentOS7需要搭配el7，而CentOS8需要搭配el8
2. Erlang版本要和RabbitMQ的版本对应
  
  RabbitMQ与Erlang的版本对照表
  
  [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-WY3f0Rp4-1687272442369)(D:/%E7%94%A8%E6%88%B7/ghp/Pictures/Typora/image-20230222164934404.png)]
Step2：下载RabbitMQ 3.8.8

Githut上的下载地址

温馨提示：可以通过Tags选择历史版本，或者最新版本
Step3：将下载完成的 Erlang 和 RabbitMQ 传输到Linux中

下载完成：

传输完成：（一般都是传输再/usr/locla目录下，而我是传输在/usr/local/src目录下）
Step4：安装socat 插件

因为RabbitMQ需要使用到这个插件，Socat 是 Linux 下的一个多功能的网络工具，名字来由是「Socket CAT」。其功能与有瑞士军刀之称的 Netcat 类似，可以看做是 Netcat 的加强版。
```
yum install socat -y
```
Step5：解压 Erlang 和 RabbitMQ

1）解压Erlang

首先要进入Erlang压缩包所在目录，Step3中放置的目录/usr/local/src
```
# 解压
rpm -ivh erlang-23.3.4.11-1.el7.x86_64.rpm
# 解压成功后，查看erlang的版本，检测是否解压成功
erl -v
```
备注：i是install的缩写，vh表示查看下载进度

2）解压RabbitMQ

首先要进入RabbitMQ压缩包所在目录，Step3中放置的目录/usr/local/src
```
rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm
```

Step6：启动RabbitMQ

# 启动RabbitMQ
systemctl start rabbitmq-server
# 关闭RabbitMQ
systemctl stop rabbitmq-server
# 查看服务状态
systemctl status rabbitmq-server

Step7：安装RabbitMQ客户端

默认情况下，RabbiMQ 没有安装 Web 端的客户端软件，需要安装才可以生效
```
# 安装RabbitMQ客户端
rabbitmq-plugins enable rabbitmq_management
# 重启RabbitMQ
systemctl restart rabbitmq-server
```
安装完毕后，重启服务，就可以看到RabbitMQ的客户端界面了
Step8：访问RabbitMQ

在Windows中的浏览器输入http://ip:15672/即可访问到RabbitMQ的客户端页面

备注：这里的 ip 是你Linux的IP，可以使用ifconfig命令查看。RabbitMQ的默认端口号为15672，默认账号和密码是guest

==注意：==Windows中访问虚拟机上的Linux，需要关闭Linux的防火墙；如果不关闭就需要开发RabbitMQ的端口号，不然无法成功访问。这里我是直接关闭了Linux的防火墙
```
# 关闭防火墙
systemctl stop firewalld.service
```
Step9：创建账号
角色固定有四种级别：
- administrator：可以登录控制台、查看所有信息、并对rabbitmq进行管理
- monToring：监控者；登录控制台，查看所有信息
- policymaker：策略制定者；登录控制台指定策略
- managment：普通管理员；登录控制
默认的账号密码仅限于本机 localhost 进行访问，所以需要添加一个远程登录的用户
```
# 查看当前所有的用户
rabbitmqctl list_users
# 创建账号和密码
rabbitmqctl add_user 用户名 密码
# 设置用户角色
rabbitmqctl set_user_tags 用户名 角色
# 为用户添加资源权限，添加配置、写、读权限
rabbitmqctl set_permissions -p "/" 用户名 ".*" ".*" ".*"
```
现在使用admin和123就可以成功登录了，登录成功后会来到下面这个页面

3、RabbitMQ初体验

编码流程可以参考前面那张RabbitMQ原理图。这里主要是演示一下使用RabbitMQ实现一个简单消息队列，实现消息的生产和消费

示例：

主要演示一下RabbitMQ的简单模式，详细代码请参考博主的Gitee仓库或Github仓库

Step1：搭建环境

1）开启RabbitMQ

2）创建Maven工程

Step2：导入依赖

        <!--rabbitmq客户端对应的依赖-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.8.0</version>
        </dependency>
        <!--操作文件流的一个依赖-->
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.6</version>
        </dependency>

Step3：编写消息生产者

package com.hhxy.rabbitmq.one;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author ghp
 * @date 2023/2/22
 * @title 生产者
 * @description
 */
public class Producer {

    // 队列名称
    public static final String QUEUE_NAME = "hello";

    // 发送消息
    public static void main(String[] args) throws Exception {
        // 1、创建一个连接工厂对象，用于创建连接
        ConnectionFactory factory = new ConnectionFactory();
        // 2、配置连接信息
        factory.setHost("192.168.88.136"); // 设置工厂IP，用于连接RabbitMQ
        factory.setUsername("admin"); // 设置用户名
        factory.setPassword("123"); // 设置密码
        // 3、创建连接（这一步需要抛异常，比如IP对应的RabbitMQ不存在或者说密码账号错误）
        Connection connection = factory.newConnection();
        // 4、获取信道
        Channel channel = connection.createChannel();
        // 5、创建队列（这里直接采用了默认的交换机，所以不需要创建交换机）
        /*
            1. 队列名称
            2. 队列中的消息是否持久化（磁盘），默认取值为false，表示不持久化，此时消息存储在内存中
            3. 队列是否排他，true表示只能同一个连接中的信道使用，false表示不同连接的信道都可以使用该队列
            4. 是否自动删除，true表示当所有消费者与该队列断开了连接，队列会自动删除
            5. 其它参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 6、发送消息
        String message = "Hello World!"; // 要发送的消息
        /*
            1. 指定要发送的交换机，空表示使用默认的交换机
            2. 指定将消息存放到哪一个队列
            3. 其它参数
            4. 指定本次要发送的消息
         */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

        System.out.println("消息发送完毕！");
    }
}

注意：这里一定要设置队列非排他，否则生产者无法接收队列中的消息，因为消费者和生产者的Connection对象都是new出来的，不是同一个连接对象

运行结果：

Step5：编写消息消费者

package com.hhxy.rabbitmq.one;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author ghp
 * @date 2023/2/22
 * @title 消费者
 * @description
 */
public class Consumer {

    // 队列名称
    public static final String QUEUE_NAME = "hello";

    // 接收消息
    public static void main(String[] args) throws Exception {
        // 1、创建一个连接工厂对象，用于创建连接
        ConnectionFactory factory = new ConnectionFactory();
        // 2、配置连接信息
        factory.setHost("192.168.88.136"); // 设置工厂IP，用于连接RabbitMQ
        factory.setUsername("admin"); // 设置用户名
        factory.setPassword("123"); // 设置密码
        // 3、创建连接（这一步需要抛异常，比如IP对应的RabbitMQ不存在或者说密码账号错误）
        Connection connection = factory.newConnection();
        // 4、获取信道
        Channel channel = connection.createChannel();
        // 5、接收消息
        /*
            1. 指定消费哪一个队列中的消息
            2. 消息接收成功后是否自动应答，true表示自动应答，false表示手动应答
            3. 消息接收成功时的回调
            4. 取消消息接收时的回调（可以理解为消息接收失败时的回调）
         */
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            // 消息成功消费后执行的逻辑（每成功消费一条消息都会执行一次这段逻辑）
            System.out.println("消息消费成功 "+message);
            // System.out.println("消息消费成功 "+new String(message.getBody()));
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            // 消息消费中断后执行的逻辑（每中断消费一条消息都会执行一次这段逻辑）
            System.out.println("消息消费中断");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

运行结果：

4、工作队列

工作队列（Work Queue），也被称为任务队列（Task Queue），是一种常用的消息队列模式，它用于将任务分发给多个消费者并进行处理。它可以很好地解决应用程序中需要处理大量耗时任务的问题。

在工作队列模式中，任务由生产者发布到队列中。然后，多个消费者可以从队列中接收任务，并且处理这些任务。任务通常是独立的、离散的单元，每个任务可以分配给一个或多个消费者进行处理。消费者在完成任务后，将任务标记为已处理，并将结果返回到另一个队列或直接发送给生产者。

工作队列模式通常具有以下特点：
- 任务可以并行处理。
- 任务可以被分配给多个消费者进行处理。
- 每个任务只能被一个消费者处理。
- 消费者可以动态地加入或退出队列。
工作队列的应用：工作队列模式在分布式系统和大规模数据处理中应用广泛，例如在Web应用程序中，将请求发送到队列中，并使用多个工作进程处理请求。这可以提高应用程序的可伸缩性和性能。常见的工作队列系统包括RabbitMQ和Apache Kafka等。

4.1 消息应答

4.1.1 消息应答简介

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息，后续发送给该消费者的消息都会发生丢失，这是十分严重的！

为了保证消息在发送过程中不丢失，引入消息应答机制，消息应答就是：消费者在接收到消息并且处理该消息之后，告诉 RabbitMQ它已经处理了，RabbitMQ才会把该消息删除（其实这个有点类似与三次握手机制）

消息应答的类别
- 自动应答：是RabbitMQ默认的配置，消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡。它没有额外的工作量，有更高的吞吐量（只要消费者能够跟上），但是存在消息丢失和消息积压的问题。如果消息队列一段时间突然接收大大量的消息，由于消费者消费不及时，以致于大量消息在消息队列中的堆积，从而导致内存耗尽；如果消息在传递到消息队列后发生了阻塞，此时由于消息发送后会被立即认为是发送成功，所以消费者中的连接或信道可能发生关闭，此时消息就无法被真正地消费，从而导致消息的丢失。
  
  综上所诉，使用自动应答只适合在消费者消费消息高效，且消费者能保持稳定的消费速率的情况下使用。
  
  前面RabbitMQ的入门案例种就是使用的自动应答模式
- 手动应答：需要自己配置，需要在消费者捕获异常，并手动确认应答状态，是ack还是nack，但是不存在消息丢失和消息积压的问题
  - Channel.basicAck (肯定确认应答)：RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了
```
// 第一个参数是消息的标记
// 第二个参数表示是否应用于多消息, true表示应用于多消息
basicAck(long deliveryTag, boolean multiple);
```
    备注：批量应答虽然效率很高，但不建议使用批量应答（也就是建议multiple设置为false），因为在批量应答的过程中容易发生消息丢失，在传输一些不太重要的但数量特别多消息时，可以使用批量应答
  - Channel.basicReject (拒绝确认应答)：
```
// 第一个参数表示拒绝，deliveryTag为对应的消息
// 第二个参数表示是否重新加入队列。true表示重新入队列，false表示丢弃或者进入死信队列
basicReject(long deliveryTag, boolean requeue);
```
    备注：该方法reject后，该消费者还是会消费到该条被reject的消息（当requeue为true时）
  - Channel.basicNack (否定确认应答)：表示己拒绝处理该消息，可以将其丢弃了
```
// 第一个参数表示拒绝,deliveryTag为对应的消息
// 第二个参数是表示否应用于多消息, true表示应用于多消息
// 第三个参数表示是否重新加入队列
basicNack(long deliveryTag, boolean multiple, boolean requeue);
```
    与 basicReject 区别就是同时支持多个消息，可以拒绝接收该消费者先前接收未 ack 的所有消息。拒绝接收后的消息也会被自己消费到
  - Channel.basicRecover：是否恢复消息到队列
```
// requeue是否重新加入队列，true 则重新入队列，并且尽可能的将之前 recover 
//的消息投递给其他消费者消费，而不是自己再次消费。false 则消息会重新被投递给自己
basicRecover(boolean requeue);
```
消息自动重新入队：如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

4.1.2 消息手动应答

默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答

示例：

准备一个生产者，两个消费者，一个消费者延迟一秒消费消息，另一个消费者延迟三十秒消费消息。

详细代码请参考博主的Gitee仓库或Github仓库

Step1：环境搭建

具体参考【3、RabbitMQ初体验】
Step2：编写生产者
Step3：编写消费者
Step4：测试

以上这个过程成功模拟了前面那张消息重新入队的动图

4.2 队列和消息的持久化

4.2.1 队列持久化

队列的持久化

前面我们演示的队列都没有进行持久化，存储在RabbitMQ中的队列默认都是没有进行持久化的，这就意味着一旦RabbitMQ服务器发生宕机，则存储在RabbitMQ中的队列包括队列中的消息都会丢失

需要注意的是，原本不持久化的队列，如果想要重新设置持久化，必须先删除（死信队列的routingkey设置也是一样的），否则会报错，错误如下所示：

删除后，再进行持久化，然后就能看到这个图标

4.2.2 消息持久化

消息的持久化

队列持久化的目的是防止RabbitMQ发生宕机导致队列丢失，而消息持久化的目的是防止消费者发生异常导致消息丢失。换句话说队列持久化并不能保障消息持久化，但队列不持久化一定不能保障消息持久化，消息持久化需要：队列持久化+消息持久化

消息持久化的属性是：MessageProperties.PERSISTENT_TEXT_PLAIN

注意：将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。

4.3 消息分发

4.3.1 不公平分发

为什么要使用不公平分发？

RabbitMQ默认的消息分发策略的轮询策略（也就是平均分发，也称公平分发），在【4.1.2 消息手动应答】当中已经演示过了。这种分发策略虽然很公平，但是公平和效率是两个对立统一的关系，公平不能保障效率，效率不能顾及公平。因为不同的消费者效率可能不同，效率高的消费者一下就将消息消费了，而效率低的消费者会耗时很久才消费消息，如果使用轮询策略，当生产者一下生产出大量的消息，会出现高效率的消费者游刃有余，而低效率的消费者发生消息堆积，甚至出现内存溢出的情况

示例：

创建一个生产者，两个消费者，一个消费者延迟1秒消费消息，另一个消费者30秒消费消息，然后生产者生产8条消息。

详细代码请参考博主的Gitee仓库或者Github仓库

Step1：搭建环境
Step2：创建生产者
Step3：创建消费者

prefetchCount默认取值为0，取值1时表示不公平开发，取值大于1时，表示预取值
Step4：测试

可以发现生产者发出8条消息，按照轮询策略，应该是消费者1消费4条消息，消费者2消费4条消息，但是由于配置了不公平分发，此时消费者1收到了7条消息，而消费者2只收到了一条消息。因为这种不公平分发机制，会优先将消息发送给处在空闲状态的消费者，当消息11发送给消费者1，消息22发送给消费者2，而消费者1延迟1秒，消费者2延迟30秒，在这30秒的时间内会讲所有请求发送给消费者1

4.3.2 预取值

预取值（perfetch）类似于Nginx中的weight属性，可以控制分发比例，比如消费者1设置的预取值为2，消费者2设置的预取值是3，那么消费者发送5条消息，消费者1会接收到两条，消费者2会接收三条。本质是消息队列与消费者通道中消息能够堆积的数量
预取值原理（个人理解，并非真正的原理）：预取值其实就是在队列和消费者的连接之间开辟了一个缓存空间，当预取值为2时，消费者会批量接收2两个消息（这个相当于设置了multiple属性，只是设置了批量数量限制），并且这个过程是和multiple过程一样，是异步的，即消息一旦被接收就会告诉RabbitMQ已处理，所以同样的会存在消息丢失问题，解决方法是限制此缓冲区的大小以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用 basic.qos 方法设置「预取计数」值来完成的

当消息缓冲区中的消息数量达到basic.qos设置的值后，RabbitMQ就会停止给该消息通道传递消息，除非至少有一个未处理的消息被确认，例如，假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK，RabbitMQ 将会感知这个情况到并再发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。

通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的 RAM（随机存取存储器）消耗，应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同 100 到 300 范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。

预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的。

综上所诉，预取值过小吞吐量会很低，预取值过大会增加RAM的消耗，一个合理的预取值将大大提高程序的性能，而合理的预取值是需要不断实验。

4.4 发布和确认

前面在4.1小节中我们学习了队列的持久化和消息的持久化，队列的持久化和消息的持久化并不能真正地保障数据的持久化，持久化是将数据保存到磁盘上，在将数据保存到磁盘的过程中，可能因为异常或者RabbitMQ宕机导致数据并没有保存在磁盘上，所以还需要关键性的一步：发布确认

PS：消息应答主要用于消费者确认已经成功处理了消息，而发布确认主要用于生产者确认消息是否已经成功保存到队列中。这两种确认机制是保证消息可靠性的重要手段，可以有效避免消息的丢失和重复消费等问题

4.4.1 发布确认概述

发布确认原理

发布确认模式（Publish-Confirm Mode）是指在使用消息队列发送消息时，通过确认机制来保证消息的可靠性，确保消息已经被队列服务器接收并存储成功。发布确认模式是一种广泛应用于消息队列的可靠消息传输机制。

在发布确认模式中，消息发送方将消息发送到队列服务器后，会等待服务器返回的确认消息。确认消息表示消息已经被服务器接收并存储成功。如果发送方在规定的时间内没有收到确认消息，就会认为消息发送失败，并进行重试或抛出异常。

发布确认模式有两种实现方式：单个确认发布和批量确认发布。单个确认发布是指每次发送一条消息后等待确认消息；批量确认发布是指每次发送多条消息后等待一次确认消息。两种方式都可以保证消息的可靠性，但批量确认发布可以提高消息发送的效率。

需要注意的是，发布确认模式对消息发送方的性能有一定的影响，因为每次发送消息都需要等待确认消息。同时，确认消息的延迟也会影响到消息的实时性，因此需要根据应用场景选择合适的发布确认模式。
单个发布确认（RabbitMQ默认是采用单个发布确认模式的）

单个确认发布（Single-Confirm Publish）是指在使用消息队列发送消息时，发送方发送一条消息后等待接收到队列服务器返回确认消息后再发送下一条消息。这种方式可以保证消息的可靠性，确保消息已经被队列服务器接收并存储，避免消息丢失或重复发送等问题。

单个确认发布模式可以有效保证消息的可靠性，但是在高并发场景下可能会出现性能问题，因为每次发送消息都需要等待确认消息。因此，在一些高性能应用中，可以使用批量确认发布（Batch-Confirm Publish）模式，即发送一批消息后等待一次确认消息，以提高发送消息的效率。
批量发布确认

批量发布确认（Batch-Confirm Publish）是指在使用消息队列发送消息时，发送方发送一批消息后等待接收到队列服务器返回的批量确认消息后再发送下一批消息。这种方式可以提高消息的发送效率，同时也可以保证消息的可靠性。

在批量发布确认模式中，消息发送方可以发送多条消息到队列服务器，然后等待服务器返回的批量确认消息。批量确认消息表示这批消息已经被服务器接收并存储成功。如果发送方在规定的时间内没有收到批量确认消息，就会认为这批消息发送失败，并进行重试或抛出异常。

批量发布确认模式可以提高消息的发送效率，因为可以一次性发送多条消息，减少了网络传输和等待确认消息的时间。同时，也可以保证消息的可靠性，因为只有当一批消息全部发送成功后，才会收到批量确认消息，避免了部分消息发送成功而另一部分消息发送失败的情况。

需要注意的是，批量发布确认模式仍然需要设置合适的超时时间来避免长时间等待确认消息而导致的性能问题。同时，由于批量发送多个消息，也需要对消息的顺序和处理逻辑进行合理的设计和考虑。当发生故障导致发布出现问题时，不知道是哪个消息出问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息
异步批量确认

异步批量确认（Asynchronous Batch-Confirm）是指在使用消息队列发送消息时，发送方发送一批消息后不需要等待接收到队列服务器的确认消息，而是异步地在后台处理确认消息。这种方式可以提高消息的发送效率和系统的吞吐量，同时也可以保证消息的可靠性。(性价比最高)

在异步批量确认模式中，消息发送方可以发送多条消息到队列服务器，然后继续发送下一批消息，而不需要等待确认消息的返回。在后台异步处理确认消息的过程中，可以统计这批消息的发送成功率，以及未发送成功的消息列表，以便进行重试或处理异常情况。

异步批量确认模式可以提高消息的发送效率和系统的吞吐量，因为不需要等待确认消息，可以在发送消息的同时进行其他操作。同时，也可以保证消息的可靠性，因为在异步处理确认消息的过程中，可以及时发现未发送成功的消息，并进行处理。

需要注意的是，异步批量确认模式需要对消息的顺序和处理逻辑进行合理的设计和考虑，确保异步处理确认消息的过程不会影响到其他业务逻辑的正常执行。同时，在异步处理确认消息的过程中，也需要设置合适的超时时间和重试机制来保证消息的可靠性。
综上所诉
- 单独发布确认：同步等待确认，简单，但吞吐量非常有限。
- 批量发布确认：批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是那条消息出现了问题。
- 异步批量确认：最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来稍微难些

4.4.2 发布确认案例

示例：

模拟单一发布确认、批量发布确认、异步批量确认，并测试三者的效率

Step1：环境搭建

参考RabbitMQ初体验，略……

Step2：编码

单个发布确认代码如下：

package com.hhxy.rabbitmq.demo04;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;

/**
 * @author ghp
 * @date 2023/2/23
 * @title
 * @description
 */
public class ConfirmMessage {

    // 消息的生产数量
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
          publishMessageIndividually(); // 单个发布确认
//        publishMessageBatch(); // 批量发布确认
//        publishMessageAsync(); // 异步批量确认
    }

    /**
     * 单个发布确认
     */
    public static void publishMessageIndividually() throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        /*
            1. 队列名称
            2. 队列中的消息是否持久化（磁盘），默认取值为false，表示不持久化，此时消息存储在内存中
            3. 队列是否排他，true表示只能同一个连接中的信道使用，false表示不同连接的信道都可以使用该队列
            4. 是否自动删除，true表示当所有消费者与该队列断开了连接，队列会自动删除
            5. 其它参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认
        channel.confirmSelect();

        long startTime = System.currentTimeMillis(); // 发送消息的开始时间

        // 批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            /*
               1. 指定要发送的交换机，空表示使用默认的交换机
               2. 指定将消息存放到哪一个队列
               3. 其它参数
               4. 指定本次要发送的消息
            */
            channel.basicPublish("", queueName, null, message.getBytes());
            // 获取确认消息(true表示消息写入磁盘成功，false表示写入磁盘失败)
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息持久化成功！");
            }
        }

        long endTime = System.currentTimeMillis(); // 发送消息的结束时间

        System.out.println("发布" + MESSAGE_COUNT + "条消息耗时: " + (endTime - startTime) + "ms");

    }

批量发布确认代码：

异步发布确认代码：

Step3：测试结果

单一发布确认：

批量发布确认：

异步批量确认：

备注：一个线程用于监听，一个线程用于发送，所以出现了上图的现象，即：消息发送完毕了，消息监听器还在执行

在前面我们使用ConfirmCallBack回调用于监听未确认的消息，但是回调是执行在发送消息之后的，并且回调函数只能拿到消息的标识，并不能处理确认的消息！

那么如何处理异步发布确认模式下，未确认的消息（也就是写入磁盘失败的消息）？

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue 这个队列在 ConfirmCallbacks 与发布线程之间进行消息的传递。

Step1：创建一个容器，用于存储未确认的消息

Step2：将所有的消息装入容器

Step3：删除容器中已经确认过的消息

Step4：未确认的消息从容器中取出来，然后重新发送

5、交换机

交换机（Exchange）是消息传递的核心组件之一

5.1 交换机概述

交换机的作用：将生产者发送的消息路由到一个或多个与之绑定的队列中，从而实现消息的传递和消费

在RabbitMQ中，生产者发送的消息需要指定一个Routing Key，Routing Key是一种标识消息的文本字符串，用于告诉RabbitMQ应该将消息路由到哪些队列。交换机根据Routing Key将消息路由到一个或多个与之绑定的队列中，这些队列上的消费者可以接收并处理这些消息
交换机的分类
- 直连交换机（Direct Exchange）：它将消息路由到与消息的Routing Key完全匹配的队列中。
- 主题交换机（Topic Exchange）：它将消息路由到与消息的主题（Topic）匹配的队列中，主题可以包含通配符（*和#）。
- 头部交换机（Headers Exchange）：它将消息路由到与消息头中指定的键值对完全匹配的队列中。
- 扇形交换机（Fanout Exchange）：它将消息路由到与该交换机绑定的所有队列中，忽略消息的Routing Key。
- 无名交换机：它是RabbitMQ默认提供的交换机，通过空字符串""进行标识，前面一直都是使用无名交换机
```
// 无名交换机
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
```
临时队列
```
// 创建一个临时队列
String queue = channel.queueDeclare().getQueue();
```
备注：队列的名称是随机的，当消费者与队列断开连接时，队列就自动删除了

创建成功后，能够在RabbitMQ的客户端上看到：
绑定(bindings)

binding 其实是 exchange 和 queue 之间的桥梁，它告诉我们 exchange 和那个队列进行了绑定关系。比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 进行了绑定

5.2 Fanout 交换机

发布订阅模式是RabbitMQ的六大核心之一，其核心实现依赖于Fanout Exchange。

个人对于发布订阅模式的理解：平常我们在使用抖音时，遇到自己觉得不错的抖音短视频，就会关注作者，然后每次关注的作者更新作品后抖音系统就会提醒我们

PS：这个功能很常见，比如B站、QQ、微信……都有这个功能

此时上面的关注操作就是订阅，之后作者发布新作品时，就会将它的新作品推广给关注它的粉丝

示例：使用 Fanout Exchange 实现发布订阅模式

创建一个生产者EmitLog，再创建两个消费者MessageReceiveLogs01和MessageReceiveLogs02，创建两个临时队列，创建一个交换机logs，交换机的类型为fanout，然后将该交换机与两个临时队列进行绑定，routingKey为空串，消费者1接收来自队列1的消息，消费者2接收来自队列2的消息。这样就实现了发布订阅模式，即：生产者每发送一条消息，交换机都能通过routingKey找到对应的队列，然后将消息传给队列，之后队列将消息发送给真在接收消息的消费者

Step1：搭建环境

还是RabbitMQ初体验的环境，略……

Step2：创建生产者

package com.hhxy.rabbitmq.demo05;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.Channel;

import java.util.Scanner;

/**
 * @author ghp
 * @date 2023/2/24
 * @title 生产者
 * @description
 */
public class EmitLog {

    // 交换机名
    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        Scanner sc = new Scanner(System.in);
        while (sc.hasNext()) {
            String message = sc.next();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println("生产者发送消息: " + message);
        }
    }
}

Step3：创建消费者

1）MessageReceiveLogs01

package com.hhxy.rabbitmq.demo05;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

import java.nio.charset.Charset;

/**
 * @author ghp
 * @date 2023/2/24
 * @title 消费者
 * @description
 */
public class MessageReceiveLogs01 {

    // 交换机的名称
    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        // 声明一个交换机
        /*
            1. 交换机的名字
            2. 交换机的类型
         */
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 声明一个临时队列
        String queueName = channel.queueDeclare().getQueue();
        // 将交换机和队列进行绑定
        /*
            1. 队列的名称
            2. 交换机的名称
            3，routingKey，绑定标识，交换机根据routingKey找到绑定的队列
         */
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println("MessageReceiveLogs01等待接收消息...");

        // 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("消息接收成功: "+new String(message.getBody()));
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息接收失败:" + consumerTag);
        };

        channel.basicConsume(queueName, true, deliverCallback, cancelCallback);


    }
}

2）MessageReceiveLogs02

代码和 MessageReceiveLogs01 类似，略……

Step4：测试

5.3 Direct 交换机

上一节中的我们的日志系统将所有消息广播给所有消费者，对此我们想做一些改变，例如我们希望将日志消息写入磁盘的程序仅接收严重错误(errros)，而不存储哪些警告(warning)或信息(info)日志消息避免浪费磁盘空间。Fanout 这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播，在这里我们将使用 direct 这种类型来进行替换，这种类型的工作方式是，消息只去到它绑定的 routingKey 队列中去。

在上面这张图中，我们可以看到 X 绑定了两个队列，绑定类型是 direct。队列 Q1 绑定键为 orange，队列 Q2 绑定键有两个：一个绑定键为 black，另一个绑定键为 green。在这种绑定情况下，生产者发布消息到 exchange 上，绑定键为 orange 的消息会被发布到队列 Q1。绑定键为 black和green 的消息会被发布到队列 Q2，其他消息类型的消息将被丢弃

当然如果 exchange 的绑定类型是direct，但是它绑定的多个队列的 key 如果都相同，在这种情况下虽然绑定类型是 direct 但是它表现的就和 fanout 有点类似了，就跟广播差不多，如上图所示

示例：

Step1：搭建环境

略……

Step2：编写生产者

package com.hhxy.rabbitmq.demo06;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.Channel;

import java.util.Scanner;

/**
 * @author ghp
 * @date 2023/2/24
 * @title 生产者
 * @description
 */
public class DirectLogs {

    // 交换机名
    public static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        Scanner sc = new Scanner(System.in);
        while (sc.hasNext()) {
            String message = sc.next();
            channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes("UTF-8"));
            System.out.println("生产者发送消息: " + message);
        }
    }
}

Step3：编写消费者

1）ReceiveLogsDirect01

package com.hhxy.rabbitmq.demo06;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

/**
 * @author ghp
 * @date 2023/2/24
 * @title 消费者
 * @description
 */
public class ReceiveLogsDirect01 {

    public static final String EXCHANGE_NAME = "direct_logs";


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        // 声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        // 声明一个队列
        channel.queueDeclare("console", false, false, false, null);
        // 将交换机和队列进行绑定
        channel.queueBind("console", EXCHANGE_NAME, "info");
        channel.queueBind("console", EXCHANGE_NAME, "warning");

        System.out.println("ReceiveLogsDirect01等待接收消息...");

        // 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("消息接收成功: " + new String(message.getBody()));
            System.out.println("交换机: " + message.getEnvelope().getExchange());
            System.out.println("路由键: " + message.getEnvelope().getRoutingKey());
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息接收失败:" + consumerTag);
        };
        channel.basicConsume("console", true, deliverCallback, cancelCallback);
    }
}

2）ReceiveLogsDirect02

略……

Step4：测试

5.4 Topic 交换机

在上一个小节中，我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机，而是使用了 direct 交换机，从而有能实现有选择性地接收日志。尽管使用 direct 交换机改进了我们的系统，但是它仍然存在局限性——比方说我们想接收的日志类型有 info.base 和 info.advantage，某个队列只想 info.base 的消息，那这个时候direct 就办不到了。这个时候就只能使用 topic 类型的交换机了

须知：

Topic交换机的RoutingKey必须符合某一种规则，即RoutingKey是一个单词列表，单词之间使用.隔开
Topic交换机的RoutingKey可以使用通配符进行匹配

1）*：可以代替一个位置

2）#：可以替代零个或多个位置
当一个队列绑定键是 #，那么这个队列将接收所有数据，就有点像 fanout 了
如果队列绑定键当中没有 # 和 * 出现，那么该队列绑定类型就是 direct 了

示例：

使用一个Topic类型交换机，实现让RoutingKey能够进行匹配转发到对应的队列中，具有三给单词，中间一个单词是 orange 的转发给Q1；具有三给单词，最后一个单词是 rabbit 的转发给Q2；已 lazy 这个单词开始的，转发给Q2

Step1：搭建环境

略……

Step2：编写生产者

package com.hhxy.rabbitmq.demo07;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.Channel;

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

/**
 * @author ghp
 * @date 2023/2/25
 * @title
 * @description
 */
public class TopicLogs {

    // 交换机名
    public static final String EXCHANGE_NAME = "topic_logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        Scanner sc = new Scanner(System.in);
        // 路由key，用于给交换机匹配队列
        Map<String, String> routingKeys = new HashMap<>();
        routingKeys.put("orange", "AA");
        routingKeys.put("orange.", "BB");
        routingKeys.put(".orange.", "CC"); // 匹配成功: Q1  *.orange.*
        routingKeys.put("others.orange.others", "DD"); // 匹配成功 Q1  *.orange.*
        routingKeys.put("others.orange.rabbit", "EE"); // 匹配成功 Q1 Q2 *.orange.*  *.*.rabbit
        routingKeys.put("others..rabbit", "FF"); // 匹配成功 Q2 *.*.rabbit
        routingKeys.put("..rabbit", "GG"); // 匹配成功 Q2 *.*.rabbit
        routingKeys.put("lazy", "HH"); // 匹配成功 Q2 lazy.#
        routingKeys.put("lazy.others", "II"); // 匹配成功 Q2 lazy.#
        routingKeys.put("lazy.others.", "JJ"); // 匹配成功 Q2 lazy.#
        routingKeys.put("lazy....", "KK"); // 匹配成功 Q2 lazy.#
        routingKeys.put("lazy.orange.rabbit", "LL"); // 匹配成功 Q2  lazy.#  *.*.rabbit
        routingKeys.put("quick.orange.orange.rabbit", "MM"); // 匹配成功 Q1 Q2  *.orange.*  lazy.#

        // 发送消息
        while (sc.hasNext()) {
            String routingKey = sc.next();
            String message = sc.next();
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
            System.out.println("生产者发送消息: " + message);
        }
    }
}

Step3：编写消费者

1）ReceiveLogsTopic01

package com.hhxy.rabbitmq.demo07;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

/**
 * @author ghp
 * @date 2023/2/25
 * @title 消费者
 * @description
 */
public class ReceiveLogsTopic01 {

    public static final String EXCHANGE_NAME = "topic_logs";


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        // 声明一个交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        // 声明一个队列
        channel.queueDeclare("Q1", false, false, false, null);
        // 将交换机和队列进行绑定
        channel.queueBind("Q1", EXCHANGE_NAME, "*.orange.*");

        System.out.println("ReceiveLogsTopic01等待接收消息...");

        // 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("消息接收成功: " + new String(message.getBody(), "UTF-8"));
            System.out.println("交换机: "+ message.getEnvelope().getExchange());
            System.out.println("路由键: " + message.getEnvelope().getRoutingKey());
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息接收失败:" + consumerTag);
        };
        channel.basicConsume("Q1", true, deliverCallback, cancelCallback);
    }
}

2）ReceiveLogsTopic012

略……

Step4：测试

6、死信队列

6.1 死信队列概述

什么是死信？
死信（Dead-Letter）顾名思义就是“死掉”的消息，一些消息因为某种原因而导致没有被正常消费调的消息，RabbitMQ中消息变成死信的原因如下：
1. 消息被拒绝：当消费者拒绝一条消息时，它会被标记为“拒绝”，并发送到死信队列中。
2. 消息过期：当一条消息的过期时间到达时，它会被标记为“过期”，并发送到死信队列中。
3. 队列长度限制：当一个队列达到了最大长度限制时，新的消息将无法被投递，而被标记为“过期”并发送到死信队列中。
4. 消息被删除：当一条消息被消费者手动删除时，它会被标记为“删除”，并发送到死信队列中。
什么是死信队列？

死信队列（Dead-Letter Queue，简称DLQ）是RabbitMQ中一个重要的概念，它是一种特殊的队列，用于接收无法被正常处理的消息。当消息被拒绝、超时或达到重试次数等情况时，它会被发送到死信队列中，供开发人员进行处理和分析。

具体来说，当一个队列中的消息无法被消费者正常处理时，RabbitMQ会将这些消息发送到一个特定的交换机中，这个交换机称为“死信交换机”。开发人员可以通过配置将这个交换机绑定到一个死信队列中，用于接收这些无法被正常处理的消息。这些消息可以被开发人员进行分析，找出出现问题的原因，进一步修复和优化系统。

简而言之，死信队列就是一个专门用来存储那些因某种原因而导致消费失败的消息。
死信队列的作用：
- 避免消息丢失：当消息无法被正常处理时，它可以被发送到死信队列中，从而避免了消息的丢失。
- 重试消息处理：死信队列可以对无法被正常处理的消息进行重新投递，从而提高消息处理的成功率。
- 分析和调试问题：通过查看死信队列中的消息，开发人员可以找出系统中出现问题的原因，并进行调试和分析。
- 延迟消息处理：死信队列可以支持消息的延迟处理，从而帮助系统更好地处理高峰期的消息处理压力
死信队列的优缺点
- 优点：
  - 提高系统可靠性：死信队列可以确保消息不会被丢失，当消息无法被正常处理时，它可以被发送到死信队列中，从而避免了消息的丢失。
  - 方便调试和分析：通过查看死信队列中的消息，开发人员可以找出系统中出现问题的原因，并进行调试和分析。
  - 支持消息重试：死信队列可以对无法被正常处理的消息进行重新投递，从而提高消息处理的成功率。
  - 提高系统的可扩展性：死信队列可以支持消息的延迟处理，从而帮助系统更好地处理高峰期的消息处理压力
- 缺点：
  - 需要额外的配置和管理：死信队列需要进行额外的配置和管理，包括创建死信队列、设置死信交换机等。
  - 可能会导致消息处理延迟：当消息被发送到死信队列中时，可能会导致消息处理的延迟，从而降低系统的响应速度。
  - 可能会占用额外的存储空间：由于死信队列需要存储无法被正常处理的消息，因此可能会占用额外的存储空间

6.2 死信队列案例

模拟死信队列的场景：

消息TTL（Time To Live，生存时间）过期
死信最大长度
死信消息被拒

示例1：模拟消息TTL过期而变成死信

Step1：搭建环境

略……

Step2：创建生产者

package com.hhxy.rabbitmq.demo08;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;

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

/**
 * @author ghp
 * @date 2023/2/25
 * @title
 * @description
 */
public class Producer {

    // 普通交换机
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    // 死信交换机
    public static final String DEAD_EXCHANGE = "dead_exchange";
    // 普通队列
    public static final String NORMAL_QUEUE = "normal_queue";
    // 死信队列
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {
        // 1、获取信道
        Channel channel = RabbitMqUtil.getChannel();

        // 2、声明交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 3、声明队列
        // 配置其它参数，用于将普通队列绑定一个死信交换机，这样就能够将普通队列中的死信转发到死信队列中
        Map<String, Object> arguments = new HashMap<>();
        // 过期时间10s，普通消息过了10s，还未被消费就变成了死信（可以直接有生产者指定）
//        arguments.put("x-dead-letter-exchange", 10000);
        // 为正常队列设置死信交换机，用于将普通队列中的死信转发到死信队列中（固定写法）
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信交换机的RoutingKey
        arguments.put("x-dead-letter-routing-key", "dead");
        // 声明一个普通队列
        channel.queueDeclare(NORMAL_QUEUE,false, false, false, arguments);
        // 声明一个死信队列
        channel.queueDeclare(DEAD_QUEUE,false, false, false, null);

        // 4、绑定交换机和队列
        // 绑定普通对列和普通交换机
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "normal");
        // 绑定死信队列和死信交换机
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "dead");


        // 5、发送消息
        // 设置消息的过期时间10s，如果普通消息经过10s还没有被消息就会添加到死信队列中
        AMQP.BasicProperties  properties =
                new AMQP.BasicProperties()
                        .builder().expiration("10000").build();
        for (int i = 0; i < 10; i++) {
            String message = "消息" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "normal", properties, message.getBytes());
        }
        System.out.println("消息发送完毕!");
    }
}

Step3：创建消费者

1）Consumer01：消费正常消息

package com.hhxy.rabbitmq.demo08;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.*;

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

/**
 * @author ghp
 * @date 2023/2/25
 * @title 消费者
 * @description
 */
public class Consumer01 {
    
    // 普通队列
    public static final String NORMAL_QUEUE = "normal_queue";

    public static void main(String[] args) throws Exception {
        // 1、获取信道
        Channel channel = RabbitMqUtil.getChannel();

        // 5、接收消息
        System.out.println("Consumer01等待接收普通消息...");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("普通消息消费成功: "+new String(message.getBody()));
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("普通消息消费失败!");
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, cancelCallback);
    }

}

备注：1、到5、中省略了消息队列和交换机的声明，这里可以声明页可以不声明，需要注意的是声明了，就一定要先启动消费者，不然会直接会因为找不到队列而报错

2）Consumer02：专门用于消费死信队列中的消息

package com.hhxy.rabbitmq.demo08;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @author ghp
 * @date 2023/2/25
 * @title 消费者
 * @description
 */
public class Consumer02 {

    // 死信队列
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception{
        // 1、获取信道
        Channel channel = RabbitMqUtil.getChannel();

        // 5、接收消息
        System.out.println("Consumer02等待接收死信消息...");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("死信消息消费成功: "+new String(message.getBody()));
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("死信消息消费失败!");
        };
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, cancelCallback);
    }

}

Step4：测试

生产者生产10条消息

消费者1消费10条消息

关闭消费者1，生产者生产10条消息，这10条消息因为TTL过期（超过10s）被死信交换机转发到了死信队列

消费者2消费死信队列中的死信消息

示例2：模拟正常队列达到最大长度，无法进入正常队列的正常消息变为死信

代码和示例1类似，只有生产者的代码不太一样，主要改动了两个地方

第一，在生产者的代码中，需要配置一下正常队列的最大长度

        // 设置正常队列能接收消息的最大数量
        arguments.put("x-max-length", 5);

设置完能够在客户端中看到Lim

第二，在生产者的代码中，去掉TTL的设置

注意：因为参数改变了，所以需要把原先队列删除

测试结果：

示例3：模拟正常信消息被拒变为死信

代码和示例2的代码类似，改动的代码也不是很多，

一是改动生产者代码，去掉正常队列的长度限制那行代码；

二是改动消费者1的代码，设置手动应答（只有手动应答才能够指定拒绝哪些消息），其次就是配置手动应答的代码，如下所示：

注意：因为参数改变了，所以需要把原先队列删除，否则会报错

测试结果：

7、延迟队列

7.1 延迟队列概述

什么是延迟队列？

延迟队列是一种在一定时间后自动将消息投递给消费者的队列。在RabbitMQ中，延迟队列通常是通过死信队列和队列消息的过期时间来实现的
延迟队列的实现流程
- Step1：创建一个普通队列，并设置一个死信交换机和一个死信路由键
- Step2：设置队列的消息过期时间，使得消息在一定时间后会变成死信消息
- Step3：将普通队列绑定到死信交换机上，同时指定一个死信路由键
- Step4：创建一个消费者来消费死信队列中的消息
常用的实现方式有：Redis的zset，Java的DelayQueue，Quartz获取Kafka的时间轮
延迟队列的优缺点
- 优点
  - 提高系统的可靠性：延迟队列可以防止消息丢失，并保证消息在指定时间后被处理，从而提高系统的可靠性。
  - 提高系统的可扩展性：延迟队列可以分担系统的负载，从而提高系统的可扩展性。
  - 支持多种应用场景：延迟队列可以实现很多应用场景，比如订单超时未支付、消息通知和任务调度等。
  - 灵活性强：延迟队列可以根据具体的业务需求进行配置，支持不同的延迟时间和消息处理方式，从而提高灵活性。
- 缺点
  - 延迟时间不准确：由于网络延迟等因素，延迟队列中的消息并不能保证在指定的时间精确地被处理。
  - 消息堆积：如果延迟时间设置过长，延迟队列中可能会积累大量的消息，导致队列堆积和消息丢失等问题。
  - 系统复杂度增加：延迟队列需要额外的配置和维护工作，增加了系统的复杂度。
延迟队列的应用场景
- 订单超时未支付：可以通过设置订单的超时时间，在订单超时后将订单信息发送到延迟队列中，以便之后对超时但未支付的订单进行处理
- 延迟发送消息：可以通过设置消息的过期时间，在指定时间后将消息发送到延迟队列中，以便在一定时间后进行消息通知
- 定时任务：可以通过设置任务的执行时间，在指定时间后将任务发送到延迟队列中，以便在指定时间执行任务
TTL的两种设置（延迟队列主要靠队列、消息的过期时间来实现）
- 消息设置TTL
```
rabbitTemplate.converAndSend("X","XC",message,correlationData -> {
    correlationData.getMessageProperties().setExpiration("5000");
});
```
- 队列设置TTL
```
Map<String, Object> params = new HashMap<>();
params.put("x-message-ttl",5000);
return QueueBuilder.durable("QA").withArguments(args).build(); // QA 队列的最大存活时间位 5000 毫秒
```
两种设置的区别

消息设置TTL，消息即使过期也不会被马上丢弃，因为消息是否过期所在即将发送给消费者之前判定的，过期就会被丢弃如果有死信队列，就添加到死信队列中；但如果队列设置TTL，队列过期就会自动删除

注意：消息或队列没有设置TTL，则默认是永不过期，如果TTL设置为0，表示此时如果不能直接转发给指定的消费者就直接丢弃或者进入死信队列

7.2 基于死信队列实现延迟队列

实例：

7.1 小节中已经讲过了，延迟队列需要TTL+死信队列，现在我们将使用SpringBoot整合RabbitMQ实现延迟队列，生产者发送一条消息，交换机 xExchange，消息会进入一个TTL为10s的消息队列A，同时进入一个TTL为60s的消息队列B，10s后过期，会被死信交换机yExchange转发给死信队列C，最终被消费者消费，同样的60s后，队列B中的消息也会变成死信，从而被消费者消费。

前面没有使用SpringBoot时，只有两个类：生产者类和消费者类，这样虽然可以实现，但是代码的耦合性太高了（后期维护成本很高，同时不利于代码的复用和统一管理），而使用SpringBoot后，我们可以将队列、交换机的声明，以及交换机和队列的绑定单独抽取出来，变成一个配置类，这样消费者就只需要生产消息，消费者只需要消费消息，这样不仅降低了代码的耦合度，也满足了单一职责原则，大大提高了系统的可维护性，总之就是使用SpringBoot具有很多好处。

Step1：搭建环境

1）创建一个SpringBoot工程

2）导入依赖

        <!--web环境-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!--fastjson依赖-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.23</version>
        </dependency>

        <!--lombok依赖-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

        <!--RabbitMQ对应的starter-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>

        <!--测试环境-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!--swagger依赖-->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>2.9.2</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>2.9.2</version>
        </dependency>

3）编写配置文件

# 端口号配置
server:
  port: 8888

# spring相关配置
spring:
  # rabbitmq相关配置
  rabbitmq:
    host: 192.168.88.136
    port: 5672
    username: admin
    password: 123

Step2：编写配置类

1）Swagger配置类

PS：这个在本次案例中并没有什么用，可以先不创建，主要是后面会用来测试

package com.hhxy.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;

/**
 * @author ghp
 * @date 2023/2/25
 * @title Swagger配置类
 * @description
 */
@Configuration
@EnableSwagger2
public class SwaggerConfig {

    /**
     * 创建API文档
     * @return 返回Swagger的实例
     */
    @Bean
    public Docket webApiConfig() {
        return new Docket(DocumentationType.SWAGGER_2)
                .groupName("webApi")
                .apiInfo(webApiInfo())
                .select()
                .build();
    }

    /**
     * API文档相关信息
     */
    private ApiInfo webApiInfo() {
        return new ApiInfoBuilder()
                .title("rabbitmq 接口文档")
                .description("本文档描述了 rabbitmq 微服务接口定义")
                .version("1.0")
                .contact(new Contact("ghp", "https://blog.csdn.net/qq_66345100?type=lately",
                        "123@qq.com"))
                .build();
    }
}

2）队列配置类：主要用来声明队列和交换机，还需要用来绑定队列和交换机

package com.hhxy.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

/**
 * @author ghp
 * @date 2023/2/25
 * @title
 * @description
 */
@Configuration
public class TtlQueueConfig {

    // 用于声明死信交换机(特定写法)
    public static final String DEAD_LETTER_EXCHANGE = "x-dead-letter-exchange";
    // 用于声明死信交换机的路由键(特定写法)
    public static final String X_DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

    public static final String X_MESSAGE_TTL = "x-message-ttl";

    // 普通交换机
    public static final String X_EXCHANGE = "X";
    // 死信交换机
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    // 普通队列
    public static final String A_QUEUE = "A";
    public static final String B_QUEUE = "B";
    // 死信队列
    public static final String C_DEAD_LETTER_QUEUE = "C";

    /**
     * 声明普通交换机 X
     */
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    /**
     * 声明死信交换机 Y
     */
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    /**
     * 声明普通队列 A
     */
    @Bean("A")
    public Queue queueA() {
        // 配置队列的参数
        Map<String, Object> arguments = new HashMap<>();
        // 配置死信交换机
        arguments.put(DEAD_LETTER_EXCHANGE, Y_DEAD_LETTER_EXCHANGE);
        // 为死信交换机配置路由键
        arguments.put(X_DEAD_LETTER_ROUTING_KEY, "CY");
        // 为 A 队列设置TTL 10s
        arguments.put(X_MESSAGE_TTL, 10000);

        return QueueBuilder.durable(A_QUEUE).withArguments(arguments).build();
    }

    /**
     * 声明普通队列 B
     */
    @Bean("B")
    public Queue queueB() {
        // 配置队列的参数
        Map<String, Object> arguments = new HashMap<>();
        // 配置死信交换机
        arguments.put(DEAD_LETTER_EXCHANGE, Y_DEAD_LETTER_EXCHANGE);
        // 为死信交换机配置路由键
        arguments.put(X_DEAD_LETTER_ROUTING_KEY, "CY");
        // 为 B 队列设置TTL 60s
        arguments.put(X_MESSAGE_TTL, 60000);

        return QueueBuilder.durable(B_QUEUE).withArguments(arguments).build();
    }

    /**
     * 声明死信队列 C
     */
    @Bean("C")
    public Queue queueC() {
        return QueueBuilder.durable(C_DEAD_LETTER_QUEUE).build();
    }

    /**
     * 队列 A 绑定交换机 X
     */
    @Bean
    public Binding aQueueBindingX(@Qualifier("A") Queue aQueue, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(aQueue).to(xExchange).with("AX");
    }

    /**
     * 队列 B 绑定交换机 X
     */
    @Bean
    public Binding bQueueBindingX(@Qualifier("B") Queue aQueue, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(aQueue).to(xExchange).with("BX");
    }

    /**
     * 队列 A 绑定交换机 Y
     */
    @Bean
    public Binding aQueueBindingY(@Qualifier("A") Queue aQueue, @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(aQueue).to(yExchange).with("AY");
    }

    /**
     * 队列 B 绑定交换机 Y
     */
    @Bean
    public Binding bQueueBindingY(@Qualifier("B") Queue bQueue, @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(bQueue).to(yExchange).with("BY");
    }

    /**
     * 队列 C 绑定交换机 Y
     */
    @Bean
    public Binding cQueueBindingY(@Qualifier("C") Queue cQueue, @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(cQueue).to(yExchange).with("CY");
    }

}

注意：修改了配置，一定要删除原来的队列，然后再重启程序，重新创建队列

Step3：编写生产者

package com.hhxy.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

/**
 * @author ghp
 * @date 2023/2/25
 * @title 生产者
 * @description 用于发送延迟消息
 */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     * @param message
     */
    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message){
        log.info("当前时间:{}, 发送一条消息给队列A和队列B:{}", new Date(), message);
        // 消息发送给X交换机，指定交换机的RoutingKey为 AX，该消息会发送给A队列，消息为第三给参数
        rabbitTemplate.convertAndSend("X", "AX", "来自ttl为10s的队列消息: "+message);
        // 消息发送给X交换机，指定交换机的RoutingKey为 BX，该消息会发送给B队列，消息为第三给参数
        rabbitTemplate.convertAndSend("X", "BX", "来自ttl为60s的队列消息: "+message);
    }
}

Step4：编写消费者

package com.hhxy.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;


/**
 * @author ghp
 * @date 2023/2/25
 * @title
 * @description
 */
@Slf4j
@Component
public class DeadLetterQueueConsumer {

    /**
     * 消费来自队列 D 的消息
     */
    @RabbitListener(queues = "C")
    public void receiveC(Message message, Channel channel) throws Exception{
        String msg = new String(message.getBody());
        log.info("当前时间:{}, 收到死信队列的消息:{}", new Date(), msg);
    }
}

Step5：测试

程序启动初识状态

生产者生产消息，使用浏览器访问链接：http://localhost:8888/ttl/sendMsg/hello

此时，队列A和队列B中会有一条消息：

10s后队列A中的消息变成死信队列，60s后队列B中的消息也会变成死信然后被yExchange交换机转发到死信队列C中：

7.3 延迟队列优化

在7.2小节中，我们通过SpringBoot整合RabbitMQ，创建了一个简单的延迟队列，虽然能够实现延迟队列的一些基本的功能，但是仍然存在许多的问题，比如：每新增一个时间需求，就需要新增一个队列。所以本小节主要对这个问题进行优化

示例：

在前面代码的基础上，我们可以添加队列，该队列的延迟时间并不是实现编码写死的，而是通过生产者指定，也就是说通过来自前端的请求决定该队列要延迟多久，从而大大提高延迟队列的灵活性

相较于前面的代码，改动之处有两给地方：一是队列配置类，一是Controller层要新增一个发消息的方法

TtlQueueController：

    /**
     * 声明一个普通队列 T（不直接指定TTL）
     */
    @Bean("T")
    public Queue queueT(){
        // 配置队列的参数
        Map<String, Object> arguments = new HashMap<>();
        // 配置死信交换机
        arguments.put(DEAD_LETTER_EXCHANGE, Y_DEAD_LETTER_EXCHANGE);
        // 为死信交换机配置路由键
        arguments.put(X_DEAD_LETTER_ROUTING_KEY, "CY");
        return QueueBuilder.durable(T_QUEUE).withArguments(arguments).build();
    }
    /**
     * 队列 T 绑定交换机 X
     */
    @Bean
    public Binding tQueueBindingX(@Qualifier("T") Queue tQueue, @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(tQueue).to(xExchange).with("TX");
    }
    /**
     * 队列 T 绑定交换机 Y
     */
    @Bean
    public Binding tQueueBindingY(@Qualifier("T") Queue tQueue, @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(tQueue).to(yExchange).with("TY");
    }

SendMessageController：

    /**
     * 发送消息（TTL由请求给出）
     * @param message
     * @param ttlTime
     */
    @GetMapping("/sendMsg/{ttlTime}/{message}")
    public void sendMsg(@PathVariable("ttlTime") String ttlTime,
                        @PathVariable("message") String message) {
        log.info("当前时间:{}, 发送一条时长是{}毫秒TTL信息给队列T:{}", new Date(), ttlTime, message);
        rabbitTemplate.convertAndSend("X", "TX", message, msg -> {
            // 设置消息的TTL
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        });
    }

测试结果：

访问链接http://localhost:8888/ttl/sendMsg/60000/message1发送第一条消息

访问链接http://localhost:8888/ttl/sendMsg/600/message2发送第二条消息

由于延迟队列是基于死信队列实现的，而队列具有一个特点先进先出，这就容易出现一些不合理的情况，比如上面发送的两条消息，第一条消息有效期是1分钟，第二条消息是600ms，按道理来讲应该是600ms的消息要先被消费的，但是由于队列的特点导致明明很短有效期的消息被前面很长有效期的消息给耽误了，导致过了一分多种才被消费，这样很容易出现消息堆积问题，所以本次优化并不完善，这是队列的特点决定是（因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行），所以我们需要找寻另一种方式实现延迟队列，那就是接下来要讲的基于插件实现延迟队列

PS：前面7.1小节中也介绍过在消息属性上设置 TTL 的方式，消息可能并不会按时「死亡」

7.4 基于插件实现延迟队列

RabbitMQ延迟插件的安装

Step1：下载RabbitMQ延迟插件
```
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.8.0/rabbitmq_delayed_message_exchange-3.8.0.ez
```
官方下载地址：Community Plugins — RabbitMQ

备注：本人安装的3.8.0版本，如果想安装历史版本可以通过Tag进行查找
Step2：传输到Linux中（放到cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins目录）
RabbitMQ和延迟插件默认位于下面的目录（我的RabbitMQ版本为3.8.8 ）
```
# RabbitMQ 安装目录
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8   
# RabbitMQ 的 plgins 所在目录
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
```

Step3：安装

# 安装
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# 安装后重启服务
systemctl restart rabbitmq-server

Step4：测试

重启RabbitMQ服务后，访问客户端，在新建交换机的选项中，如果可以看到 x-delayed-message选项，就说明插件安装成功了

示例：

之前基于死信，延迟的是地方在于死信队列，而基于插件延迟的地方在交换机，如下图所示：

基于插件实现延迟队列案例代码示意图：

PS：延迟队列和死信队列并不是因为本身而成为延迟队列的，而是由于交换机才变成延迟队列和死信队列的，核心是交换机

Step1：搭建环境

略……

Step2：编写配置类

package com.hhxy.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

/**
 * @author ghp
 * @date 2023/2/26
 * @title 延迟队列配置类（基于插件实现延迟队列）
 * @description
 */
@Configuration
public class DelayedQueueConfig {
    // 延迟队列
    public static final String DELAYED_QUEUE = "delayed.queue";
    // 延迟交换机
    public static final String DELAYED_EXCHANGE = "delayed.exchange";
    // 交换机的路由键
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
    // 延迟交换机的类型（延迟插件提供的固定写法，专门用来设置延迟交换机）
    public static final String DELAYED_EXCHANGE_TYPE = "x-delayed-message";


    /**
     * 声明延迟队列
     */
    @Bean
    public Queue delayedQueue() {
        return new Queue(DELAYED_QUEUE);
    }


    /**
     * 声明延迟交换机
     */
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> arguments = new HashMap<>();
        // 设置延迟交换机的的路由方式
        arguments.put("x-delayed-type", "direct");
        /*
            1. 交换机的名称
            2. 交换机的类型
            3. 交换机是否持久化，true表示持久化
            4. 交换机是否需要自动删除，true表示需要自动删除
            5. 其它参数
         */
        return new CustomExchange(DELAYED_EXCHANGE, DELAYED_EXCHANGE_TYPE, true, false, arguments);
    }

    /**
     * 延迟队列 delayedQueue 绑定延迟交换机 delayedExchange
     */
    @Bean
    public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,
                                                      @Qualifier("delayedExchange") CustomExchange delayedExchange) {
        return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
    }

}

Step3：编写生产者

在SendMessageController类中添加下面方法

    /**
     * 发送消息（基于延迟插件实现的延迟队列）
     */
    @GetMapping("/sendMsg2/{ttlTime}/{message}")
    public void sendMsg2(@PathVariable("ttlTime") Integer ttlTime,
                        @PathVariable("message") String message) {
        log.info("当前时间:{}, 发送一条时长是{}毫秒TTL信息给延迟队列delayed.queue:{}", new Date(), ttlTime, message);
        rabbitTemplate.convertAndSend("delayed.exchange", "delayed.routingkey", message, msg -> {
            // 设置消息的延迟时间（单位ms）
            msg.getMessageProperties().setDelay(ttlTime);
            return msg;
        });
    }

Step4：编写消费者

package com.hhxy.consumer;

import com.hhxy.config.DelayedQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.security.auth.login.Configuration;
import java.util.Date;

/**
 * @author ghp
 * @date 2023/2/26
 * @title
 * @description 消费来自延迟队列的延迟消息（基于延迟插件实现延迟队列）
 */
@Slf4j
@Component
public class DeadQueueConsumer {

    @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE)
    public void receiveDelayedQueue(Message message){
        String msg = new String(message.getBody());
        log.info("当前时间:{}, 收到死信队列的消息:{}", new Date(), msg);
    }
}

Step5：测试

访问链接http://localhost:8888/ttl/sendMsg2/60000/message1发送第一条消息

访问链接http://localhost:8888/ttl/sendMsg2/600/message2发送第二条消息

8、发布确认高级

8.1 发布确认高级介绍

前面（第7节）我们的示例都是在理想状态（RabbitMQ不会宕机）上进行的，对于一些极端状态上面的示例是会出现错误的，比如：我们发送一条消息给交换机，结果该交换机不存在，那么该消息就直接丢失了；又或者说我们发送消息，有交换机，但是交换机路由的队列不存在，此时消息也会丢失，这两种情况都是由于在消息发送时，RabbitMQ出现了异常而导致的

备注：前面我们的发布确认是针对消费者的，现在我们的发布确认是针对交换机的

前面我们学习了发布确认，但那种只是对消费者进行一个监测，我们是无法对交换机和队列的状态进行监测的！发布确认是对前面的一个补充，让生产者能够监测到交换机和队列是否接收到消息
- 监测交换机实现步骤：
  
  原始状态，消息未发送到交换机，就直接丢失了；
  
  监测状态，消息未发送到交换机，能够让生产者监测到交换机是否接收到消息
  - Step1：开启发布和确认模式
```
 publisher-confirm-type: correlated
```
  - Step2：编写一个MyCallBack类，实现RabbitTemplate.ConfirmCallback接口
    1. 需要重写confirm方法，该方法会在是交换机接收消息后的回调
    2. 将接口实现类MyCallBack注入到RabbitTemplate中，让RabbitTemplate能够调用实现类的 confirm 方法
      
      注入代码：rabbitTemplate.setConfirmCallback(this);
- 监测队列实现步骤：
  - Step1：开启消息回退
```
        publisher-returns: true
        template:
            mandatory: true
```
    PS：mandatory参数也可以不设置，也能生效，但是一般都是加上的，原因如下：
    1. spring.rabbitmq.template.mandatory属性的优先级高于spring.rabbitmq.publisher-returns的优先级
    2. spring.rabbitmq.template.mandatory属性可能会返回三种值null、false、true,
    3. spring.rabbitmq.template.mandatory结果为true、false时会忽略掉spring.rabbitmq.publisher-returns属性的值
    4. spring.rabbitmq.template.mandatory结果为null（即不配置）时结果由spring.rabbitmq.publisher-returns确定
```
// 代码中开启消息回退
rabbitTemplate.setMandatory(true);
```
  - Step2：编写一个MyCallBack类，实现RabbitTemplate.ReturnsCallback接口
    1. 需要重写returnedMessage方法，该方法会在是消息回退后执行（也就是RoutingKey不可达时）
    2. 将接口实现类MyCallBack注入到RabbitTemplate中，让RabbitTemplate能够调用实现类的 returnedMessage方法
      
      注入代码：rabbitTemplate.setReturnsCallback(this);

8.2 发布确认高级案例

示例：

本案例中，主要模拟消息无法发送给交换机，交换机的RoutingKey不可达。

Step1：搭建环境

1）创建SpringBoot工程

工程目录：

2）导入依赖，和延迟队列的依赖一样，请参考7.2，略……

3）编写配置文件
```
# 端口号配置
server:
  port: 8888

# spring相关配置
spring:
  # rabbitmq相关配置
  rabbitmq:
    host: 192.168.88.136
    port: 5672
    username: admin
    password: 123
    publisher-confirm-type: correlated # 开启发布确认模式
```
publisher-confirm-type具有三种取值
- NONE 值是禁用发布确认模式，是默认值
- CORRELATED 值是发布消息成功到交换器后会触发回调方法
- SIMPLE 值经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker

Step2：编写配置类

1）ConfirmConfig：用于声明队列、交换机，以及绑定队列和交换机

package com.hhxy.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author ghp
 * @date 2023/2/26
 * @title
 * @description
 */
@Slf4j
@Configuration
public class ConfirmConfig {

    // 交换机
    public static final String CONFIRM_EXCHANGE = "confirm.exchagne";
    // 队列
    public static final String CONFIRM_QUEUE = "confirm.queue";
    // 路由键
    public static final String CONFIRM_ROUTING_KEY = "key1";

    /**
     * 声明交换机
     */
    @Bean
    public DirectExchange confirmExchange() {
        return new DirectExchange(CONFIRM_EXCHANGE);
    }

    /**
     * 声明队列
     */
    @Bean
    public Queue confirmQueue() {
        return QueueBuilder.durable(CONFIRM_QUEUE).build();
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding confirmQueueBindingConfirmExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                                      @Qualifier("confirmExchange") DirectExchange confirmExchange) {
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    }
}

2）MyCallBack：配置两个方法，一个用于处理无法到达交换机的消息，一个用于处理RoutingKey不可达的消息

package com.hhxy.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * @author ghp
 * @date 2023/2/26
 * @title 回调接口
 * @description 消息发送到交换机后的回调
 */
@Slf4j
@Configuration
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 将实现类注入到到接口中（不然通过RabbitTemplate只能调用到ConfirmCallback，而不是它的实现类）
    @PostConstruct // 被注解的方法，在对象加载完依赖注入后执行(在其它注解执行完后再执行，防止空指针异常)
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * 交换机接收到消息后执行的回调
     *
     * @param correlationData 保存消息的ID和相关数据
     * @param ack             交换机是否成功接收消息的结果，true表示交换机接收到消息
     * @param cause           交换机成功接收消息为 null，未接收到消息，返回失败的原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 获取消息id（三目表达式防止空指针异常）
        String id = correlationData != null ? correlationData.getId() : "";
        if (ack) {
            // 交换机成功接收消息
            log.info("交换机成功收到id为{}的消息", id);
        } else {
            log.info("交换机还未收到id为{}的消息, 原因:{}", id, cause);
        }
    }

    /**
     * 当消息无法路由的时候的回调方法
     * message      消息
     * replyCode    编码
     * replyText    退回原因
     * exchange     从哪个交换机退回
     * routingKey   通过哪个路由 key 退回
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.error("消息{}, 被交换机{}退回, 退回原因:{}, 路由key:{}",
                new String(returned.getMessage().getBody()), returned.getExchange(),
                returned.getReplyText(), returned.getRoutingKey());
    }

}

Step3：编写生产者

package com.hhxy.controller;

import com.hhxy.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author ghp
 * @date 2023/2/26
 * @title
 * @description
 */
@Slf4j
@RestController
@RequestMapping("/confirm")
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     */
    @GetMapping("/sendMessage/{message}")
    public void SendMessage(@PathVariable String message){
        log.info("发送消息:{}", message);
//        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUTING_KEY, message);
        CorrelationData correlationData = new CorrelationData("1"); // 设置消息ID
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUTING_KEY, message, correlationData);

    }
}

Step4：编写消费者

package com.hhxy.consumer;

import com.hhxy.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author ghp
 * @date 2023/2/26
 * @title
 * @description
 */
@Slf4j
@Component
public class Consumer {
    @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)
    public void receiveConfirmMessage(Message message){
        log.info("消费者成功接收消息:{}", new String(message.getBody()));
    }
}

Step5：测试

交换机接收消息成功，执行成功：

模拟交换机失效，无法接收消息（直接在生产者发送消息时，故意将交换机的名字写错)：

PS：添加了confirm方法用于处理消息无法发送给交换机的问题后，如果不加这个方法时，消息会直接丢失

模拟队列出现故障，无法接收消息（直接故意将RoutingKey写错）

添加了returnedMessage方法专门用于处理RoutingKey不可达的消息

9、备份交换机

第8节，我们通过confirm和returnedMessage两个方法，获得了对无法投递消息的感知能力，但通过这两个方法我们还是无法完善地处理这些消息，最多打个日志，然后触发报警，再来手动处理。而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢？

前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。在 RabbitMQ 中，有一种备份交换机的机制存在，可以很好的应对这个问题。

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

示例：

代码和8、中的类似，通过配置一个备份交换机，就能够处理路由不可达的消息了，，重构方向如下图所示

测试：

对于路由不可达的消息，是会有备份交换机转发到专门存储这些消息的队列的，保障了消息不被丢失，同时也会有一个专门的消费者来处理这些消息

注意：当我们同时配置了消息回退、备份交换机，经过测试可以发现，备份交换机的优先级更高（消息不会回退，而是直接被备份交换机转发到备份队列中）。

10、其它队列

10.1 幂等性

什么是幂等性？

幂等性是指对同一个操作的多次执行，产生的结果与单次执行的结果相同。在计算机系统中，由于各种原因（如网络不稳定、系统故障等），可能会出现重复执行同一操作的情况，如果操作本身是幂等的，就可以避免由于重复执行而导致的数据错误或不一致等问题
幂等性有什么作用？
- 避免重复操作：在分布式系统中，由于网络不稳定或其他原因，可能会导致某个操作被多次执行，如果该操作是幂等的，可以避免由于重复操作导致的数据错误或不一致等问题。
- 提高系统可靠性：在面对各种异常情况时，如果系统具有幂等性，可以确保系统状态的正确性和一致性，保障系统的可靠性和稳定性。
- 支持事务性操作：在数据库操作中，如果数据库支持幂等性操作，可以避免由于事务重复提交或回滚等问题导致的数据错误或不一致等问题。
- 提高系统的性能：由于幂等性操作可以重复执行而不产生任何影响，因此可以在分布式系统中进行负载均衡和故障恢复，提高系统的性能和可扩展性
常见的幂等性操作

在实际开发中，常见的一些操作是幂等的，例如：
- 插入操作：如果插入的数据已经存在，再次插入不会产生新的数据，不会改变系统状态。
- 删除操作：如果删除的数据已经不存在，再次删除也不会产生新的数据，不会改变系统状态。
- 更新操作：如果更新的数据已经是最新状态，再次更新不会产生新的数据，不会改变系统状态。
注意：幂等性并不是所有操作都具备的特性，有些操作可能会对系统状态产生影响，不能重复执行。在开发中，需要根据具体的业务需求和系统设计，判断哪些操作需要具备幂等性，对于不具备幂等性的操作，需要采取其他措施来确保系统的正确性和可靠性

eg：举个简单的例子吧，以前玩天天酷跑，在抽奖时，再点击抽奖的瞬间，我们可以通过快速关闭网络，从而卡刷星抽奖次数（可以无限刷，当然后来这个Bug被修改了），这就是由于抽奖这个操作没有保障幂等性导致的，在网络环境差的情况下，重复操作不能保障数据的一致性；再比如我们在购物时，我们在点击支付时，支付扣款成功，但是返回结果的时候网络异常，此时钱已经扣了，用户再次点击按钮，此时会进行第二次扣款，返回结果成功，用户查询余额发现重复扣钱了，流水记录也变成了两条
为什么会出现上面例子(eg)中的问题呢？

消费者在消费 MQ 中的消息时，MQ 已把消息发送给消费者，消费者在给 MQ 返回 ack 时网络中断，故 MQ 未收到确认信息，该条消息会重新发给其他的消费者，或者在网络重连后再次发送给该消费者，但实际上该消费者已成功消费了该条消息，造成消费者消费了重复的消息，也就会出现能够卡无限抽奖、下单多次扣钱的情况了。所以可想而知，幂等性是多么的重要！
如何实现幂等性？

MQ 消费者的幂等性的解决一般使用全局 ID 或者写个唯一标识比如：时间戳或者 UUID 或者 Redis的setnx命令或者订单消费者消费 MQ 中的消息也可利用 MQ 的该 id 来判断，或者可按自己的规则生成一个全局唯一 id，每次消费消息时用该 id 先判断该消息是否已消费过。

主流的有两种：唯一ID+指纹码机制、利用Redis命令的原子性实现

10.2 优先级队列

什么是优先级队列？

优先级队列是一种特殊的队列，它可以根据元素的优先级进行排序，优先级高的元素先出队列。在优先级队列中，每个元素都有一个与之关联的优先级，通常是一个数字或者其他类型的标识
优先级队列的应用场景有哪些？
- 任务调度：如果有多个任务需要执行，可以根据任务的优先级进行排序，优先执行优先级高的任务。
- 消息传递：如果有多个消息需要传递，可以根据消息的优先级进行排序，优先传递优先级高的消息。
- 缓存管理：如果有多个数据需要缓存，可以根据数据的优先级进行排序，优先缓存优先级高的数据。
优先级队列可以用各种数据结构实现，例如堆、红黑树、链表等。在实际开发中，需要根据具体的需求和场景，选择合适的数据结构和算法实现优先级队列，以达到最优的性能和效果。

需要注意的是，优先级队列虽然可以提高系统的效率和性能，但是也可能会导致一些问题，例如优先级反转、优先级饥饿等问题，需要根据具体的应用场景和需求，进行适当的优化和改进。
什么是优先级反转？

优先级反转是指在优先级队列中，一个优先级较低的任务或消息阻塞了一个优先级较高的任务或消息的执行，导致整个系统的性能下降的现象。

例如，假设有一个任务调度系统，其中有两个任务，分别是优先级高的任务 A 和优先级低的任务 B。在任务队列中，任务 B 在任务 A 前面排队等待执行。但是由于任务 B 的执行时间比较长，它一直阻塞着任务 A 的执行，导致任务 A 的优先级无法得到充分发挥，整个系统的性能下降。

优先级反转是一种比较常见的问题，需要在实际开发中进行适当的优化和改进。一种常用的方法是使用优先级反转避免算法，即当一个低优先级的任务占用了一个高优先级任务所需要的资源时，系统会暂停低优先级任务的执行，等待高优先级任务执行完成后再继续执行低优先级任务。另外，还可以使用动态优先级调整等方法来避免优先级反转问题。
什么是优先级饥饿？

优先级饥饿是指在优先级队列中，优先级低的任务或消息始终得不到执行的现象，导致它们一直等待在队列中，而优先级高的任务或消息则得到优先执行的机会。

例如，假设有一个任务调度系统，其中有多个任务，分别具有不同的优先级。在任务队列中，优先级高的任务总是排在前面等待执行，而优先级低的任务则一直等待在队列的后面，始终得不到执行的机会。这种情况下，优先级低的任务就会出现优先级饥饿的问题。

优先级饥饿是一种比较常见的问题，通常需要在实际开发中采取一些措施来避免。一种常用的方法是使用公平调度算法，即在任务队列中为不同优先级的任务分配合适的执行时间，以确保每个任务都有机会得到执行。另外，还可以使用动态调整优先级等方法来解决优先级饥饿问题。
声明优先级队列的方式
- 方式一：可视化操作（通过RabbitMQ的Web客户端新增一个优先级队列）
  
  注意：优先级排序标号最大可以设置到 255，官网推荐 10 左右如果设置太高比较吃内存和 CPU（标号越大，优先级越高)
- 方式二：编码实现
  - Step1：声明一个优先级队列（这一步一般在配置类中编写）
```
Map<String, Object> params = new HashMap();
// 设置优先级标号
params.put("x-max-priority", 10);
channel.queueDeclare("priority.queue", true, false, false, params);
```
  - Step2：为消息设置优先级（这一步一般在生产者中编写）
```
AMQP.BasicProperties properties = new AMQP.BasicProperties()
    									 .builder().priority(5).build();
```
⚠注意事项：不能一开始就开启消费者，需要先开启生产者，等到消息到达队列中才开启消费者，因为消息进入优先级队列需要一定的时间进行排序（当然这个时间很短，自己预估一下）

设置成功后：

示例：

创建一个优先级队列，然后创建一个消费者，一个生产者，生产者发送10条消息，第5条消息设置优先级

Step1：搭建环境

和3、RabbitMQ初体验中的环境一样，略……

Step2：创建生产者

package com.hhxy.producer;


import com.hhxy.utils.RabbitMqUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

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

/**
 * @author ghp
 * @date 2023/2/27
 * @title
 * @description
 */
public class PriorityProducer {

    private static final String QUEUE_NAME = "priority.queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtil.getChannel();

        // 声明队列
        Map<String, Object> params = new HashMap<>();
        // 设置优先级队列最大标识（0~10）
        params.put("x-max-priority", 10);
        channel.queueDeclare(QUEUE_NAME, true, false, false, params);

        //给消息赋予一个priority属性
        AMQP.BasicProperties properties =
                new AMQP.BasicProperties().builder().priority(1).priority(10).build();

        for (int i = 0; i < 10; i++) {
            String message = "消息" + i;
            if (i == 5) {
                // 给第5条消息消息设置优先级
                channel.basicPublish("", QUEUE_NAME, properties, message.getBytes());
            } else {
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
            System.out.println("消息发送完成：" + message);
        }
    }
}

Step3：创建消费者

package com.hhxy.consumer;

import com.hhxy.utils.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;

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

/**
 * @author ghp
 * @date 2023/2/27
 * @title
 * @description
 */
@Slf4j
public class PriorityConsumer {

    private final static String QUEUE_NAME = "priority.queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtil.getChannel();

        // 发送消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            log.info("消费者PriorityConsumer接收消息:{}, 消息tag为:{}", message, consumerTag);
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            log.info("消息消费过程被中断！");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

Step4：测试

发送十条消息到优先级队列中

消费者消费消息：

10.3 惰性队列

什么是惰性队列？

RabbitMQ 从 3.6.0 版本开始引入了惰性队列的概念。惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持更多的消息存储。当消费者由于各种各样的原因(比如消费者下线、宕机亦或者是由于维护而关闭等)而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。

默认情况下，当生产者将消息发送到 RabbitMQ 的时候，队列中的消息会尽可能的存储在内存之中，这样可以更加快速的将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当 RabbitMQ 需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。虽然 RabbitMQ 的开发者们一直在升级相关的算法，但是效果始终不太理想，尤其是在消息量特别大的时候。
惰性队列的使用场景
- 避免内存溢出：长时间不能消费消息而造成堆积时，惰性队列可以尽可能地将消息存入磁盘中，而不是一直占用内存，从而避免了内存占用过高的问题。
- 避免队列溢出：当队列中的消息非常多时，惰性队列可以支持更多的消息存储，从而避免了队列溢出的问题。
- 避免队列阻塞：当RabbitMQ需要释放内存时，惰性队列可以将内存中的消息换页至磁盘中，从而避免了队列被阻塞的问题
惰性队列的两种模式

队列具备两种模式：default 和 lazy。默认的为 default 模式，在 3.6.0 之前的版本无需做任何变更。lazy 模式即为惰性队列的模式，可以通过调用 channel.queueDeclare 方法的时候在参数中设置，也可以通过 Policy 的方式设置，如果一个队列同时使用这两种方式设置的话，那么 Policy 的方式具备更高的优先级。如果要通过声明的方式改变已有队列的模式的话，那么只能先删除队列，然后再重新声明一个新的

两种模式的对比：

在发送 1 百万条消息，每条消息大概占 1KB 的情况下，普通队列占用内存是 1.2GB，而惰性队列仅仅占用 1.5MB

惰性队列的实现方式

方式一：可视化创建（在RabbitMQ的Web客户端操作）

方式二：编码创建（使用x-queue-mode参数）

Map<String, Object> arguments = new HashMap<String, Object>();
// 设置惰性队列
args.put("x-queue-mode", "lazy");
channel.queueDeclare("myqueue", false, false, false, arguments);

11、MQ集群

RabbitMQ总共有三种模式
- 单一模式：即单机情况不做集群，就单独运行一个 RabbitMQ
- 普通模式：默认模式，以两个节点（A、B）为例来进行说明：
  - 当消息进入 A 节点的 Queue 后，Consumer 从 B 节点消费时，RabbitMQ 会在 A 和 B 之间创建临时通道进行消息传输，把 A 中的消息实体取出并经过通过交给 B 发送给 Consumer。
  - 当 A 故障后，B 就无法取到 A 节点中未消费的消息实体；如果做了消息持久化，那么得等 A 节点恢复，然后才可被消费；如果没有持久化的话，就会产生消息丢失的现象。
- 镜像模式 ：经典的 Mirror 镜像模式，保证数据不丢失：
  - 高可靠性解决方案，主要就是实现数据的同步，一般来讲是 2 - 3 个节点实现数据同步。
  - 对于 100% 数据可靠性解决方案，一般是采用 3 个节点。
  - 在实际工作中也是用得最多的，并且实现非常的简单，一般互联网大厂都会构建这种镜像集群模式
另外，还有主备模式，远程模式，多活模式等等

11.1 普通模式

示例：普通模式

准备三台服务器（在虚拟机上开启三给Linux），它们的hostname的名字分别为：node1、node2、node3，node1的ip地址是192.168.88.136，node2的ip地址是192.168.88.137，node3的ip地址是192.168.88.138

相关Linux指令

ifconfig # 查看ip
hostname # 查看hostname
vi /etc/hostname # 修改hostname，注意需要root权限
vi /etc/hosts # 配置节点ip
reboot # 重启Linux
sync # 将内存数据保存到磁盘中

注意：

修改完hostname后需要重启Linux
每次重启Linux都会刷新IP
vi /etc/hosts指令要为每一台服务器都需要配置三个节点的IP
```
192.168.88.136 node1
192.168.88.137 node2
192.168.88.138 node3
```

Step1：前期准备

1）准备三台服务器

2）配置hostname，将三台服务器配置为node1、node2、node3，配置完后记得重启一下Linux

3）配置hosts，需要同时配置三个节点的IP

Step2：在节点1执行

确保每个节点的 Cookie文件实用的是同一个值

# 让节点1和节点2的Cookie文件进行关联
scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/.erlang.cookie
# 让节点1和节点3的Cookie文件进行关联
scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/.erlang.cookie

Step3：重启、

启动 RabbitMQ 服务，顺带启动 Erlang 虚拟机和 RbbitMQ 应用服务
```
# 每台服务器都需要重启一遍
rabbitmq-server -detached
```
注意：每台服务器都需要重启一遍

Step4：在节点2执行

# 关闭RabbitMQ服务器
rabbitmqctl stop_app
# 重置服务器
rabbitmqctl reset
# 将2号节点加入1号节点
rabbitmqctl join_cluster rabbit@node1
# 启动RabbitMQ服务器
rabbitmqctl start_app

备注：rabbitmqctl stop 会将 Erlang 虚拟机关闭，rabbitmqctl stop_app 只关闭 RabbitMQ 服务

Step5：在节点3执行

# 关闭RabbitMQ服务器
rabbitmqctl stop_app 
# 重置服务器
rabbitmqctl reset
# 将3号节点加入1号节点
rabbitmqctl join_cluster rabbit@node2
# 启动RabbitMQ服务器
rabbitmqctl start_app

备注：如果执行了Step4，将2号节点加入到了1号节点中，其实这里，也可以直接将3号节点加入到2号节点中

Step6：在节点1执行

# 重启node1
rabbitmqctl start_app
# 查看集群状态
rabbitmqctl cluster_status

Step7：测试

搭建完集群后，需要重新创建账号
```
# 创建账号并设置密码
rabbitmqctl add_user admin 123
# 设置用户角色
rabbitmqctl set_user_tags admin administrator
# 设置用户全新啊
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
```
访问http://192.168.88.136:15672，登录RabbitMQ的Web客户端，然后如果看到以下图片，就说明集群搭建成功了

再次强调：一定要记得件三台服务器的防火墙给关了（不然又要浪费好多时间取排错🤣）

Step8：解除集群（分别在node2和node3中执行）

如果集群中有一个节点宕机了，就可以通过以下命令将其脱离出来

# 停止服务
rabbitmqctl stop_app
# 重置服务
rabbitmqctl reset
# 启动服务
rabbitmqctl start_app
# 查看机器装填
rabbitmqctl cluster_status
# 断开node1和node2（这条指令只在node1上运行）
rabbitmqctl forget_cluster_node rabbit@node2

11.2 镜像模式

如果 RabbitMQ 集群中只有一个 Broker 节点，那么该节点的失效将导致整体服务的临时性不可用，并且也可能会导致消息的丢失。可以将所有消息都设置为持久化，并且对应队列的durable属性也设置为true，但是这样仍然无法避免由于缓存导致的问题：因为消息在发送之后和被写入磁盘井执行刷盘动作之间存在一个短暂却会产生问题的时间窗。通过 publisherconfirm 机制能够确保客户端知道哪些消息己经存入磁盘，尽管如此，一般不希望遇到因单点故障导致的服务不可用。

引入镜像队列(Mirror Queue)的机制，可以将队列镜像到集群中的其他 Broker 节点之上，如果集群中的一个节点失效了，队列能自动地切换到镜像中的另一个节点上以保证服务的可用性。

综上所诉，使用普通模式，节点无法复用，即：node1上创建的队列，如果node1宕机，则node1上创建的队列也会失效，此时其它节点或者消费者访问node1上的队列，就会无法访问，node1中的消息也会随着宕机而丢失

Step1：启动三台服务器上的RabbitMQ

Step2：设置备份

ha-mode	ha-params	Desc
all	忽略	all表示镜像到集群上的所有节点，ha-params参数忽略
exactly	节点数量	exactly表示镜像到设置数量的节点，ha-params节点数量
nodes	节点列表	nodes表示镜像到指定节点列表上，ha-params节点列表

备注：按照上方这样设置，主机和备机这两个节点之一的一个挂掉，都会再重新再复制一个，总能够保障有一个节点上能包含挂掉节点上的队列

Step3：测试

直接使用之前的代码，创建一个队列，然后发送一条消息（注意队列要添加mirriro）

package com.hhxy.rabbitmq.demo01;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @author ghp
 * @date 2023/2/22
 * @title 生产者
 * @description
 */
public class Producer {

    // 队列名称
    public static final String QUEUE_NAME = "mirrior.hello";

    // 发送消息
    public static void main(String[] args) throws Exception {
        // 1、创建一个连接工厂对象，用于创建连接
        ConnectionFactory factory = new ConnectionFactory();
        // 2、配置连接信息
        factory.setHost("192.168.88.136"); // 设置工厂IP，用于连接RabbitMQ
        factory.setUsername("admin"); // 设置用户名
        factory.setPassword("123"); // 设置密码
        // 3、创建连接（这一步需要抛异常，比如IP对应的RabbitMQ不存在或者说密码账号错误）
        Connection connection = factory.newConnection();
        // 4、获取信道
        Channel channel = connection.createChannel();
        // 5、创建队列（这里直接采用了默认的交换机，所以不需要创建交换机）
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 6、发送消息
        String message = "Hello World!"; // 要发送的消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("消息发送完毕！");
    }
}

测试成功，镜像创建成功了，现在每次在node1上创建一个队列，在node2上也会创建一个

注意：镜像模式中，当node1发生宕机时，RabbitMQ会自动将node2上的队列复制到node3上（也就是说他会自动备份，保障任何时刻都有一个节点具有其它宕机节点上的队列），这样能够保障集群中至少有一个节点能够具有访问的队列

之前使用普通模式只要node1宕机，node1上所有的队列也就跟着无法访问了，现在即使node1宕机，node1上的队列仍然能够被访问，同时消息也没有发生丢失

11.3 实现集群的高可用

11.2小节我们通过镜像队列保障了宕机节点上的队列能够被访问，从而提高了高可用性，但是仍然存在一个比较大的不足之处。那就是我们的代码中ip地址是写死的，如果我们代码中那个ip地址对应的节点宕机了，那就无法访问节点了！所以我们需要进行负载均衡配置。主要利用Haproxy+keepalive进行代理

备注：VIP是Haproxy提供的代理IP，任何发送给VIP的请求都会被 Haproxy 接收，然后Haproxy再将这个请求转发给MQ集群，这样放置了前面出现的 IP 写死的问题，一般情况 VIP 的请求都会发送给主机（node11），如果node11发生了宕机，keepalive（一款软件）会监测到，然后将发送给主机的请求转发给备机，从而大大提高了高可用性

11.4 实现消息的可靠传递和数据的同步

备注：感觉这个现在学了用不上，长时间不用也会很容易忘，所以为了节约时间我就直接大致看一遍，就没有仔细学了，等以后要用的时候再来仔细学习一下

11.4.1 Federation

Federation是什么？

Federation 是 RabbitMQ 提供的一种插件，用于将多个 RabbitMQ 集群连接起来形成一个逻辑上的整体。Federation 可以使得不同的 RabbitMQ 集群之间能够进行消息的跨集群传递和交换，从而实现跨集群的消息路由和数据同步。

Federation 插件的核心功能是支持 AMQP 0-9-1 协议中的远程队列和远程交换机，通过这些远程实体的配置和路由规则，可以实现消息的跨集群路由和数据同步。Federation 插件还提供了一些高级特性，例如支持动态发现和自动连接其他集群、消息过滤、消息的 TTL 等。

Federation 插件可以被广泛应用于分布式系统、多数据中心场景、跨地域消息传递等领域。通过使用 Federation 插件，可以将不同的 RabbitMQ 集群连接在一起，形成一个更加可靠和可扩展的消息服务体系，提高了整个系统的可用性和灵活性。
Federation的作用有哪些？
- 实现跨集群的消息路由：Federation 可以使不同的 RabbitMQ 集群之间进行消息的跨集群传递和交换，从而实现跨集群的消息路由。
- 支持多数据中心场景：在多数据中心场景下，Federation 可以将不同数据中心的 RabbitMQ 集群连接起来，实现消息的跨数据中心传递和数据同步。
- 提高系统可靠性和可用性：通过将多个 RabbitMQ 集群连接在一起，形成一个更加可靠和可扩展的消息服务体系，提高了整个系统的可用性和灵活性。
- 支持动态发现和自动连接其他集群：Federation 插件支持动态发现和自动连接其他集群，从而可以实现动态的集群连接和路由规则的更新。
- 提供消息过滤和 TTL 支持：Federation 插件还提供了消息过滤和 TTL 支持，可以根据不同的需求对消息进行过滤和控制。
FederationExchange是什么？

Federation Exchange 是 RabbitMQ 提供的一种交换机类型，它是通过 RabbitMQ 的 Federation 插件实现的。与其他类型的交换机不同，Federation Exchange 可以实现跨集群的消息路由和数据同步，从而支持分布式系统和多数据中心应用场景。

具体来说，Federation Exchange 可以将来自不同 RabbitMQ 集群的消息路由到目标集群，实现跨集群的消息传递和数据同步。在 Federated Exchange 中，交换机绑定到外部集群中的 Exchange，然后通过 Exchange 之间的桥接路由器进行跨集群消息路由。同时，Federation Exchange 还支持 TTL 和消息过滤等特性，可以根据不同的需求对消息进行过滤和控制。

总的来说，Federation Exchange 是 RabbitMQ 提供的一个强大的消息路由和数据同步解决方案，它可以帮助构建分布式系统和多数据中心应用，提高消息传递的可靠性和可用性，同时提供了丰富的特性和灵活的配置方式，使得系统更加可靠、可扩展和易于管理。
FederationExchange有什么用？
- 实现跨集群的消息路由和数据同步：Federation Exchange 可以将来自不同 RabbitMQ 集群的消息路由到目标集群，从而实现跨集群的消息传递和数据同步。这对于分布式系统和多数据中心的应用场景非常有用，可以大大提高消息传递的可靠性和可用性。
- 提高系统的可靠性和可用性：Federation Exchange 支持消息 TTL 和过滤等特性，可以根据不同的需求对消息进行过滤和控制，从而提高系统的可靠性和可用性。
- 提供丰富的特性和灵活的配置方式：Federation Exchange 提供了丰富的特性和灵活的配置方式，可以根据不同的需求进行定制，比如可以设置消息的优先级、持久化、重试等特性，可以根据路由键和队列名进行消息过滤和路由选择等。
- 支持异构系统集成：Federation Exchange 可以跨不同版本的 RabbitMQ 和不同的消息协议进行消息传递和数据同步，支持异构系统之间的集成和交互。
综上所述，Federation Exchange 是 RabbitMQ 提供的一个强大的消息路由和数据同步解决方案，它可以帮助构建分布式系统和多数据中心应用，提高消息传递的可靠性和可用性，同时提供了丰富的特性和灵活的配置方式，使得系统更加可靠、可扩展和易于管理。

举个简单的例子：

一个公司它有很多的业务，需要部署在不同的服务器上，如果一个业务部署在深圳（他有一个交换机AExchange），另一个业务部署在北京，此时深圳的程序员需要发送一条消息给AExchange，此时由于距离很近，延迟很低，即使增加了事务或者publisherconfirm机制，依旧性能会很高；但是如果北京的程序员想要发送消息给AExchange，此时由于距离很远，延迟很大，特别是添加了事务或者publisherconfirm机制，那此时回应就会很慢很慢，严重影响系统的性能。此时你可能回想，为什么不把业务全部部署在深圳呢？你要想一下，集群一般是比较大的公司才会采用的，业务也会有很多很多，如果全部放在深圳，会极大的降低容灾性，一旦发生事故后果可想而知！Federation插件很好地解决了这个问题
FederationQueue是什么？

FederationQueue 是 RabbitMQ 提供的一种队列类型，它通过 RabbitMQ 的 Federation 插件实现，主要用于实现跨集群的队列复制和数据同步，在分布式系统和多数据中心的应用场景中，FederationQueue 可以帮助实现消息的可靠传递和数据的同步

示例：

RabbitMQ默认是关闭Federation的，所以我们需要手动打开它

Step1：每个节点上执行

# 下载federation插件
rabbitmq-plugins enable rabbitmq_federation
# 开启federation
rabbitmq-plugins enable rabbitmq_federation_management

11.4.2 Shovel

备注：感觉这个现在学了用不上，长时间不用也会很容易忘，所以为了节约时间我就直接大致看一遍，就没有仔细学了，等以后要用的时候再来仔细学习一下

Shovel是什么？

Shovel 是 RabbitMQ 提供的一个插件，作用类似于Federation，用于实现消息的复制和数据的同步。它可以将来自一个 RabbitMQ 队列的消息复制到另一个队列中，并支持对消息进行过滤、转换、重命名等操作。Shovel 插件可以在不同的 RabbitMQ 服务器之间或不同的 vhost 之间进行消息复制，支持 AMQP 和 STOMP 协议。
Shovel的作用
- 实现消息的可靠传递和数据的同步：Shovel 插件可以将来自一个 RabbitMQ 队列的消息复制到另一个队列中，从而实现消息的可靠传递和数据的同步。它可以在不同的 RabbitMQ 服务器之间或不同的 vhost 之间进行消息复制，支持 AMQP 和 STOMP 协议。
- 提供丰富的配置选项：Shovel 插件提供了丰富的配置选项，可以对消息进行过滤、转换、重命名等操作，从而满足不同的业务需求。
- 支持多种消息模式：Shovel 插件支持多种消息模式，包括点对点、发布订阅、广播等模式，可以根据业务需求选择合适的模式进行消息复制和数据同步。
- 提高系统的可靠性和可用性：Shovel 插件支持队列持久化、镜像队列等特性，可以根据不同的需求对队列进行配置和管理，从而提高系统的可靠性和可用性。
综上所述，Shovel 是 RabbitMQ 提供的一个强大的消息复制和数据同步解决方案，它可以帮助构建分布式系统和多数据中心应用，提高数据的可靠性和可用性，同时提供了丰富的配置选项和多种消息模式，使得系统更加可靠、可扩展和易于管理。
Shovel和Federation的比较
- 相同点：
  - 实现消息的可靠传递和数据的同步：Shovel 和 Federation 都可以将来自一个 RabbitMQ 队列的消息复制到另一个队列中，从而实现消息的可靠传递和数据的同步。
  - 支持多种消息模式：Shovel 和 Federation 都支持多种消息模式，包括点对点、发布订阅、广播等模式。
  - 提高系统的可靠性和可用性：Shovel 和 Federation 都支持队列持久化、镜像队列等特性，可以提高系统的可靠性和可用性
- 不同点：
  - 适用场景不同：Shovel 主要适用于不同的 RabbitMQ 集群之间或不同的 vhost 之间进行消息复制和数据同步，而 Federation 则适用于将消息从一个 RabbitMQ 集群传输到另一个 RabbitMQ 集群，实现跨数据中心的数据同步。
  - 数据同步方式不同：Shovel 通过拉取方式将数据从一个队列复制到另一个队列，而 Federation 则通过推送方式将数据从一个 RabbitMQ 集群发送到另一个 RabbitMQ 集群。
  - 支持的协议不同：Shovel 支持 AMQP 和 STOMP 协议，而 Federation 仅支持 AMQP 协议。

示例：

Shovel和Federation类似，RabbitQM默认都是关闭的，需要手动开启

# 下载RabbitMQ
rabbitmq-plugins enable rabbitmq_shovel
# 开启Shovel
rabbitmq-plugins enable rabbitmq_shovel_management

总结

在这里插入图片描述

参考资料：

RabbitMQ一套通关

RabbitMQ学习笔记

[RabbitMQ]AMQP 0-9-1：模型 - 知乎 (zhihu.com)

RabbitMQ学习笔记

（总结）Nginx/LVS/HAProxy负载均衡软件的优缺点详解 (ha97.com)

RabbitMQ入门学习笔记

文章目录

RabbitMQ学习笔记

前言

1、RabbitMQ概述

1.1 MQ概述

1.2 RabbitMQ概述

2、RabbitMQ的安装

3、RabbitMQ初体验

4、工作队列

4.1 消息应答

4.1.1 消息应答简介

4.1.2 消息手动应答

4.2 队列和消息的持久化

4.2.1 队列持久化

4.2.2 消息持久化

4.3 消息分发

4.3.1 不公平分发

4.3.2 预取值

4.4 发布和确认

4.4.1 发布确认概述

4.4.2 发布确认案例

5、交换机

5.1 交换机概述

5.2 Fanout 交换机

5.3 Direct 交换机

5.4 Topic 交换机

6、死信队列

6.1 死信队列概述

6.2 死信队列案例

7、延迟队列

7.1 延迟队列概述

7.2 基于死信队列实现延迟队列

7.3 延迟队列优化

7.4 基于插件实现延迟队列

8、发布确认高级

8.1 发布确认高级介绍

8.2 发布确认高级案例

9、 备份交换机

10、其它队列

10.1 幂等性

10.2 优先级队列

10.3 惰性队列

11、MQ集群

11.1 普通模式

11.2 镜像模式

11.3 实现集群的高可用

11.4 实现消息的可靠传递和数据的同步

11.4.1 Federation

11.4.2 Shovel

总结

相关文章

9、备份交换机