消息中间件(Leo)

news/2025/10/24 10:48:03/文章来源:https://www.cnblogs.com/Leo0O-Luo/p/19162636

消息中间件

知识框架

RabbitMQ

RabbitMQ如何保证消息不丢失？

第一个是开启生产者确认机制，确保生产者的消息能到达队列，如果报错可以先记录到日志中，再去修复数据
第二个是开启持久化功能，确保消息未消费前在队列中不会丢失，其中的交换机、队列、和消息都要做持久化
第三个是开启消费者确认机制为 auto，由 spring 确认消息处理成功后完成 ack，当然也需要设置一定的重试次数，我们当时设置了 3 次，如果重试 3 次还没有收到消息，就将失败后的消息投递到异常交换机，交由人工处理

RabbitMQ 消息的重复消费问题如何解决的？

嗯，这个我们还真遇到过，是这样的，我们当时消费者是设置了自动确认机制，当服务还没来得及给 MQ 确认的时候，服务宕机了，导致服务重启之后，又消费了一次消息。这样就重复消费了因为我们当时处理的支付（订单 | 业务唯一标识），它有一个业务的唯一标识，我们再处理消息时，先到数据库查询一下，这个数据是否存在，如果不存在，说明没有处理过，这个时候就可以正常处理这个消息了。如果已经存在这个数据了，就说明消息重复消费了，我们就不需要再消费了

面试官：那你还知道其他的解决方案吗？

候选人：嗯，我想想～其实这个就是典型的幂等的问题，比如，redis 分布式锁、数据库的锁都是可以的

生产者确认机制

RabbitMQ提供了publisher confirm机制来避免消息发到MQ过程中丢失。消息到MQ以后，会返回一个结果给发送者，表示消息是否处理成功

消息失败后如何处理？

回调方法即时重发
记录日志
保存到数据库然后定时重发，成功消费后即刻删除表中的数据

消费者确认机制

RabbitMQ支持消费者确认机制，即消费者处理消息后可以向MQ发送ACK回执，MQ收到ACK回执后才删除改消息

SpringAMQP则允许三种确认模式：

manual：手动akc，需要在业务代码结束后，调用api发送ACK
auto：自动ACK，由spring监测listenter代码是否出现异常，没有异常则返回ack；抛出异常则返回nack
none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后会立即删除

可以利用Spring的retry机制，在消费者出现异常时利用本地重试，设置重试机制，当次数到达了后，如果消息依旧消费失效，将消息投递到异常交换机，交由人工处理

消息持久化

MQ默认是内存存储消息，开启持久化功能可以确保缓存在MQ的消息不丢失

持久化流程：

写入磁盘：消息到达broker(消息服务器) 后，先写入内存缓冲区（提升效率），再异步/同步刷盘至磁盘文件

同步刷盘：消息写入磁盘成功后才返回“发送成功”给生产者，可靠性高，但是性能低
异步刷盘：消息先存储内存，定期批量刷盘，性能高，但极端情况下可能丢失少量信息（刷盘前宕机）

元数据记录：同时记录信息的元信息(偏移量、主题、队列、过期时间等)，用于后续检索和恢复

消息消费后的清理机制：

消费者成功消费并发送ACK后，broker会标记消息已消费，并在合适时机从磁盘删除，避免磁盘空间占用
消费者未成功消费（处理失败），消息会保留在磁盘，等待重试或进入死信队列

对比维度	Redis	MySQL（InnoDB）	MQ（如 Kafka）
刷盘核心目标	兼顾 “缓存性能” 与 “数据可恢复”	保证 “事务 ACID”，数据绝对不丢	保证 “消息不丢失”+“高吞吐”
刷盘数据类型	全量数据（RDB）或操作日志（AOF）	redo log（事务日志）+ 数据页	消息本身（带元数据，如偏移量）
特殊优化	AOF 重写（合并冗余操作，减小文件）；RDB 压缩（节省磁盘）	双写缓冲（避免 redo log 频繁刷盘）； checkpoint（异步刷数据页）	顺序写（磁盘效率最高的写入方式）；副本同步（多副本刷盘才算成功）

死信交换机

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
消息是一个过期消息，超时无人消费
要投递的队列消息堆积满了，最早的消息可能成为死信

如果该队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，简称 DLX）。

flowchart LRA[publisher] --> B[simple.direct]B --> C[simple.queue]C --> D[consumer]C --> E[dl.direct]E --> F[dl.queue]D -- reject --> CC -- dead-letter-exchange = dl.direct<br>dead-letter-routing-key = dl --> EE -- dl --> F

TTL（Time-To-Live）

TTL，也就是 Time-To-Live。如果一个队列中的消息 TTL 结束仍未消费，则会变为死信，ttl 超时分为两种情况：

消息所在的队列设置了存活时间
消息本身设置了存活时间

flowchart LRA[publisher] -- ttl = 5000 --> B[ttl.direct]B -- ttl --> C[ttl.queue]C -- x-message-ttl = 10000 --> CC --> D[dl.direct]D -- dl --> E[dl.queue]E --> F[consumer]

延迟队列

延迟队列就是用到了死信交换机和 TTL（消息存活时间）实现的。

如果消息超时未消费就会变成死信，在 RabbitMQ 中如果消息成为死信，队列可以绑定一个死信交换机，在死信交换机上可以绑定其他队列，在我们发消息的时候可以按照需求指定 TTL 的时间，这样就实现了延迟队列的功能。

我记得 RabbitMQ 还有一种方式可以实现延迟队列，在 RabbitMQ 中安装一个死信插件，这样更方便一些，我们只需要在声明交换机的时候，指定这个就是死信交换机，然后在发送消息的时候直接指定超时时间就行了，相对与死信交换机 + TTL 要省略了一些步骤

消息堆积解决方法

当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被丢弃，这就是消息堆积问题

三种解决思路：

增加更多消费者，提高消息消费速度
在消费者内开启线程池加快消息处理速度
扩大队列容积，提高堆积上限，采用惰性队列
- 在声明队列的时候可以设置属性x-queue-mode为lazy，即惰性队列
- 基于磁盘存储，上限更高
- 性能较稳定，但基于磁盘存储，受限于磁盘IO，时校性会降低

惰性队列

惰性队列的特征：

接收到消息后直接存储到磁盘而非内存
消费者要消费消息时才会从磁盘中读取并加载到内存
支持数百万条的消息存储

RabbitMQ的高可用机制

在生产环境下采用集群保证高可用：

普通集群、镜像集群、仲裁集群

普通集群

也叫做标准集群具备下列特征：

会在集训的各个节点共享部分数据，包括：交换机，队列元信息。不包括队列中的信息；
当访问集群的某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回
队列所在节点宕机，队列中的消息就会丢失

镜像集群

本质是主从模式，具备以下特征：

交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份
创建队列的节点叫做该队列的主节点，备份到其它节点叫做该队列的镜像节点
一个队列的主节点可能是另一个队列的镜像节点
所有操作都是主节点完成，然后备份到镜像节点
主宕机后，镜像节点会替代成为新的主

仲裁队列

是3.8版本后才有的新功能，用来替代镜像队列，具备下列特征：

与镜像队列一样，都是主从模式，支持主从数据同步
使用非常简单，没有复杂的配置
主从同步基于Raft协议，强一致(解决镜像队列宕机丢数据)

RabbitMQ、RocketMQ和Kafka对比

对比维度	Kafka	RabbitMQ	RocketMQ
设计定位	高吞吐日志 / 数据同步（大数据场景）	灵活路由的通用消息队列（企业级业务）	高可靠 + 高吞吐的企业级消息队列（互联网业务）
开发语言	Scala/Java	Erlang	Java
核心优势	1. 百万级 TPS（高吞吐）2. 持久化性能强（顺序写磁盘）3. 分布式架构易扩容	1. 路由灵活（交换机类型多：Direct/Topic/Fanout/Headers）2. 社区成熟，生态完善3. 支持死信、重试、延迟等细粒度控制	1. 兼顾高吞吐（十万级 TPS）与低延迟2. 原生支持分布式事务、定时消息3. 适合复杂业务场景（如订单、支付）
核心劣势	1. 消息可靠性需依赖配置（如多副本）2. 路由功能简单（仅 Topic 路由）3. 延迟消息需二次开发	1. 高吞吐场景下性能较弱（万级 TPS）2. Erlang 语言开发，定制化难度高	1. 生态较 Kafka/RabbitMQ 弱2. 延迟消息精度较低（分钟级）
消息延迟	毫秒级（高吞吐下略高）	微秒级（轻量场景下极低）	毫秒级
可靠性保障	依赖副本数（`replication-factor`），`acks=-1` 时不丢消息	需手动配置 “队列 + 消息持久化”，支持事务消息	原生支持事务消息、消息轨迹，可靠性默认更强
社区与生态	大数据领域生态第一（适配 Spark/Flink）	通用 MQ 生态第一（客户端语言多）	阿里系生态，国内互联网企业用得多