学习和掌握RabbitMQ及其与springboot的整合实践（篇一）

前言

这篇博客主要讲述rabbitMQ的理论基础和springboot整合指南，由于篇幅过长，我将这篇博客分为两个文章讲述，本篇文章主要讲述rabbitMQ的相关概念、理论基础，篇二讲述springboot与rabbitMQ的整合实践。

相关基础概念

同步调用和异步调用

• 同步调用和异步调用是编程中处理任务执行的两种基本模式，他们的主要区别在于调用方是否需要等待被调用方的操作完成才能继续执行后续程序。
• 同步调用，调用方在调用其他服务后处于线程阻塞状态，直到被调用方返回结果，继续执行后续任务逻辑
• 异步调用，调用方在调用其他服务后，继续执行后续任务，无需等待被调用方返回结果

为什么不直接异步调用目标服务

• 值得注意的是，异步调用不是像同步调用那样，直接调用目标服务，而是将异步消息发送给消息中间件broker，或者叫消息队列，这是因为消息中间件提供了一套​​可靠的机制​​，解决了简单异步调用在分布式环境下难以处理的​​可靠性、可扩展性和系统耦合度​​等问题。

  核心问题	简单异步调用的问题	消息中间件（Broker）的解决方案
  可靠投递​	依赖进程内存，进程崩溃或重启会导致消息全部丢失	Broker将消息​​持久化存储​​在磁盘上，即使服务重启，消息也不会丢失，确保可被消费
  系统解耦​​	生产者需明确知道消费者地址，服务间直接依赖，耦合紧密，增减生产者复杂度。	生产者将消息发送至Broker即可返回，​​无需关心​​具体有多少个消费者、它们是谁或是否在线，实现了服务间的彻底解耦
  应对流量洪峰​​​	无缓冲区。瞬时高并发请求会直接压垮后端服务，导致服务雪崩。	Broker作为​​缓冲区​​，积压瞬时海量请求，后端服务可按照自身处理能力匀速消费，实现​​削峰填谷​​，保护后端系统
  应对流量洪峰​​​	一个消费者实例故障，其负责处理的任务就会丢失或阻塞，缺乏有效的重试和转移机制	Broker提供​​确认与重试机制​​。消费者处理成功后需向Broker确认；若处理失败或超时未确认，Broker会将消息重新投递给其他消费者实例，避免单点故障

消息中间件

• 消息中间件，broker，也称消息队列，MessageQueue，简单理解就是一个异步调用的消息代理软件，生产者(调用方)将异步信息发送给broker，broker将负责将消息转发给消费者(被调用方)
• 目前在项目开发中，Kafka和RabbitMQ的应用较为广泛。
• 如果项目应用场景需要​​低延迟​​、​​复杂的消息路由​​，并且强调​​消息的可靠投递​​，​​RabbitMQ​​ 是更合适的选择
• 如果项目应用场景首要目标是处理​​海量数据流​​，需要​​高吞吐量​​和​​消息持久化存储​​以便后续分析（比如收集用户行为日志进行实时分析），那么 ​​建议使用Kafka

RabbitMQ

RabbitMQ的安装部署

关于rabbitMQ的安装部署、创建虚拟主机、web控台台界面的操作使用，可以详见我的这篇博客CentOS 7 环境下 RabbitMQ 的部署与 Web 管理界面基本使用指南

RabbitMQ的整体架构与核心概念

核心概念

• publisher：消息发送方、调用方、生产者，也就是发送异步消息的一方
• exchange：交换机，负责路由消息到队列
• queue：队列，负责存储消息、转发消息到消费者
• consumer：消息接收方、被调用者、消费者，也就是接收异步消息的一方
• virtuaHost：虚拟主机，起到虚拟隔离的作用，一个RabbitMQ可以创建多个虚拟主机，在每个虚拟主机上创建交换机exchange和队列queue，作用类似于MySQL中的database，或者pg数据库中的scheme，客户端与RabbitMQ建立连接时，需要指定用户和virtuaHost，因为每个虚拟主机virtuaHost都是归属于某个或者多个用户，每个用户只对归属于自己的虚拟主机有操作权限，而每个虚拟主机上有各自的交换机exchange和队列queue配置。

整体架构（核心运转流程）

RabbitMQ的整体架构就如同上述架构图所示，结构非常清晰明了，下面主要说一下异步消息发送给RabbitMQ后，消息是如何通过RabbitMQ的内部处理，最终转发给消费者的。

1. 生产者通过指定端口与RabbitMQ建立TCP连接，并配置用户名、密码和虚拟主机
2. 生产者向RabbitMQ的指定交换机发送异步消息
3. 交换机根据绑定关系将消息路由到一个或多个关联队列
4. 消费者建立与RabbitMQ的连接，并订阅特定队列
5. 队列将异步消息推送给对应的消费者
6. 消费者处理成功后返回basic.ack响应消息，RabbitMQ 从队列中删除消息；若未确认或处理失败（返回basic.nack），消息可能重新入队或进入死信队列（DLQ）。

交换机exchange

交换机是消息的“交通枢纽”，负责接收生产者发送的消息，并根据其类型和设定的规则，将消息路由到一个或多个队列中，在我们的开发工作中，应用最多的交换机类型主要有三种：Direct （直连）、Fanout（广播） 、Topic（主题） 。

交换机类型	路由规则	核心特点	典型应用场景
​​Direct (直连)​	精确匹配​​ 消息的 ​​Routing Key​​ 和队列的 ​​Binding Key	简单高效，是默认的交换机类型	点对点精准任务分发，如将错误日志路由到专门的处理队列
​​Fanout (广播)​	忽略 Routing Key​​，将消息​​广播​​到所有与之绑定的队列	转发速度最快，纯粹的发布/订阅模型	系统全局通知、新闻推送、需要多个服务同时处理同一事件的场景
​​Topic (主题)​​	使用通配符对 ​​Routing Key​​ 进行​​模式匹配​​。* 匹配一个词，# 匹配零个或多个词	灵活性高，可以实现复杂的消息订阅机制	需要按类别订阅的场景

• Direct （直连）类型的交换机，会通过 一个binding key 与队列进行绑定，异步消息发送到Direct类型交换机时需要指定一个routing key，交换机会将异步消息精确路由到对应的队列，这个队列与交换机绑定的binding key 等于异步消息routing key
• ​​Topic (主题)​​类型的交换机，工作过程与Direct 类型的交换机类似，只不过binding key 不是一个明确的字符串，而是一个可能包含* 和 # 字符的通配字符串，举个例子，当一个 ​​Topic 类型的交换机通过com.baidu.# 这个binding key与一个队列关联时，一个routing key为 com.baidu.www会被路由到这个队列
• RabbitMQ还有一个内建的​​默认交换机（Default Exchange）​​，它是一个Direct类型的交换机。每个声明的队列都会自动以队列名作为Binding Key绑定到它上面。当你发送消息时不指定交换机，就会使用这个默认交换机

队列queue

队列，消息的存储和消费，是消息的最终目的地和缓冲区，负责将消息主动发送给消费者处理。队列类型主要从功能特性来划分。

• ​​持久化队列​​：通过设置 durable=true声明，即使RabbitMQ服务器重启，队列本身和其中的持久化消息也不会丢失，适用于重要业务数据
• 自动删除队列​​：设置 auto-delete=true后，当最后一个消费者断开连接时，队列会被自动删除，适用于临时任务
• 排他队列​​：设置 exclusive=true后，该队列仅对声明它的连接可见，连接断开时队列自动删除，常用于临时性响应队列。
• 延迟队列​​：RabbitMQ本身不直接支持，但可通过​​死信队列+消息TTL​​或官方插件 ​​rabbitmq-delayed-message-exchange​​ 实现，常用于订单超时未支付等场景。什么是延时队列，就是交换机将消息转发到队列后，队列不立刻发送给消费者，而是等待一段时间后，再发送给消费者处理。下文会详细说明什么是死信队列＋消息TTL，简单来说就是消息设置一个过期时间，时间到了这条消息就成为死信，然后转发到死信队列，消费者再从死信队列中拿到这条消息，就达到了延迟的效果。如果是安装了官方延迟插件，生产者发送消息时，在消息头（headers）中指定一个 x-delay参数（单位毫秒）来设置延迟时间。该交换机会在延迟时间到达后，才将消息路由到相应的队列
• 优先级队列​​：通过设置 x-max-priority参数，允许高优先级的消息被优先消费

MQ消息可靠性问题

消息中间件的一大特性就是有一套机制，能够保证消息的可靠性，简单说就是异步消息从生产者发出后能够顺利被消费者处理，在分布式系统中，消息队列的可靠性至关重要。RabbitMQ 作为一款流行的消息中间件，其消息可靠性需要从​​生产者​​、​​MQ 本身（Broker）​​ 和​​消费者​​三个环节共同保障。

死信队列

什么是死信

死信队列是保障消息可靠性的一个重要机制，同时也可用于实现延时消息功能。什么是死信，当一个队列中的消息满足以下任一条件时，这条异步消息就是死信

• 消费者使用basic.reject或 basic.nack返回消费失败回执，并且消息的requeue参数设置为false
• 消息是一个过期消息（达到了队列或消息本身设置的过期时间，注：队列超过设置的期间，队列要被删除，队列中的所有消息都是死信），超时无人消费
• 要投递的队列消息堆积满了，最早的消息可能成为死信

什么是死信队列

• 死信队列并非一种特殊类型的队列，它本质上就是一个普通的 RabbitMQ 队列。其特殊性仅在于它的​​用途​​——专门用来存放那些“死亡”的消息
• 死信队列其实是死信交换机和死信队列的统称，如果一个交换机是用来路由死信的，将死信路由到一个普通队列，那么这组交换机和队列统称为死信队列。

• 要让死信队列生效，我们需要对​​原始队列（即正常处理业务的队列）配置核心参数​​x-dead-letter-exchange(DLX)​​：指定当有消息成为死信时，应该将这些消息发送到哪个​​死信交换机，然后死信交换机将死信路由到死信队列，并由专门处理死信的消费者处理。

生产者可靠性

生产者可靠性的目标是确保消息能成功发送到 RabbitMQ 并被正确路由，生产者可靠性机制主要通过以下两个机制实现。

开启生产者确认机制

• 这是最关键的步骤。主要是确保消息成功转发到MQ的交换机，客户端与MQ建立连接时可以通过 publisher-confirm-type参数进行配置，配置值分为三种类型，分别是：
• none:关闭confirm机制
• simple：同步阻塞等待MQ的回执信息
• correlated：MQ异步回调方式返回回执信息
• 实际开发中一般会配置correlated类型，生产者将异步消息发送后继续执行后续逻辑，不需阻塞等待MQ的回执
• 消息成功转发到交换机，Broker 返回的确认ACK回执，投递失败则会返回NACK回执
• 在程序开发中我，我们主要通过 ConfirmCallback接口来处理这些回执，如果接收到NACK，可以重新发送或者记录日志
  

开启消息退回机制

• 即使消息成功到达交换机，也可能因路由键错误等原因无法路由到任何队列。消息退回机制就是确认消息成功从交换机路由到队列，通过设置 publisher-returns: true并实现 ReturnCallback接口，可以捕获这些路由失败的消息，以便进行后续处理，如记录错误日志或告警

MQ可靠性

MQ可靠性的目标是安全地存储消息，即使自身重启或出现故障，消息也不会丢失。MQ本身的可靠性主要通过持久化和部署高可用集群保障

全面开启持久化​

这是防止 MQ 重启导致消息丢失的核心手段。我们要确保​​交换机​​、​​队列​​和​​消息本身​​都设置为持久化

• 队列持久化​​：在声明队列时设置 durable=true。
• 交换机持久化​​：在声明交换机时设置 durable=true。
• 消息持久化​​：在发送消息时，将消息的投递模式（deliveryMode）设置为 2（持久化）。在 Spring AMQP 中，默认发出的消息就是持久化的

值得注意的是，只有设置了持久化的异步消息，在服务器夯机或者重启时，持久化队列才会将持久化消息写入磁盘保存，避免消息丢失，如果消息不是持久化，或者队列不是持久化队列，消息都会永久丢失。

部署高可用集群

• 单节点的持久化仍存在风险（如磁盘损坏）。通过部署 ​​镜像队列（Mirrored Queues）​​，可以将队列的内容复制到集群中的多个节点上。这样，即使主节点（Master）宕机，镜像节点（Slave）可以提升为主节点，继续提供服务，从而实现高可用。RabbitMQ 3.8+ 版本引入的 ​​Quorum 队列​​ 基于 Raft 协议，提供了更强的数据一致性保证，是生产环境的新推荐选择

消费者可靠性

消费者的目标是确保消息被成功处理，并且在处理失败时有恰当的补救措施，消费者可靠性保证主要通过两个方面保障，消息处理成功与否通知到位，处理失败后有补救措施、

消费者确认机制

消费者确认机制（Consumer Acknowledgement）是为了确认消费者是否成功处理消息。当消费者处理消息结束后，会向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态

• ack：成功处理消息，RabbitMQ从队列中删除该消息
• nack/reject：消息处理失败，RabbitMQ需要再次投递消息，如果消息的requeue参数设置为false，那这条消息就是死信，RabbitMQ会删除或者投递到死信队列。

如果消费者返回nack/reject，rabbitMQ默认会采用轮询的方式继续投递，会将消息投递到下一个空闲的消费者节点，怎么判断消费者是否是空闲？

• 这主要是通过消费者的预取计数（prefetchCount）参数来实现的，这个参数表示当前消费者的最大未确认消费数，举个例子，如果这个消费者配置prefetchCount为3，如果此时此刻该消费者有两条消息未返回回执，那么这个消费者在rabbitMQ看来处于空闲状态，如果有3条消息未返回回执，达到prefetchCount值，就处于繁忙状态，此时rabbitMQ不会分发消息给当前消费者。

其中，消费者确认机制可以配置为自动状态和手动状态，自动状态是一旦消费者接收到消息，处理之前就会返回回执信息，MQ接收到回执会立即从队列中删除异步消息，如果消费者处理失败，将没有补救的机会，因此，实际开发中一般设置为手动模式，在消费者成功处理之后再返回回执

实现失败重试与死信队列​

• 消费者重试​​：可以利用 Spring Retry 等机制在消费者端实现本地重试。例如，配置最大重试次数为 3 次，在发生临时性故障（如网络波动）时，可以在本地立即重试，避免频繁与 Broker 交互
• 死信队列(DLQ) ：当消息经过本地多次重试后仍然失败，不应无限次地重新投递（这可能会拖垮系统）。更优雅的做法是返回nack/reject，requeue参数设置为false， 这样这条消息就是死信，配合死信队列，由专门的消费者统一处理这些死信，这样，既不影响正常队列的消息流动，又能对失败的消息进行隔离，便于后续分析原因或进行人工干预。