news 2026/9/23 19:20:10

RabbitMQ 保证消息不丢失的几种手段

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
RabbitMQ 保证消息不丢失的几种手段

一.RabbitMQ消息丢失的三种情况

第一种:生产者弄丢了数据。生产者将数据发送到 RabbitMQ 的时候,可能数据就在半路给搞丢了,因为网络问题啥的,都有可能。

第二种:RabbitMQ 弄丢了数据。MQ还没有持久化自己挂了。

第三种:消费端弄丢了数据。刚消费到,还没处理,结果进程挂了,比如重启了。

二.RabbitMQ消息丢失解决方案

1.针对生产者

方案1 :开启RabbitMQ事务

可以选择用 RabbitMQ 提供的事务功能,就是生产者发送数据之前开启 RabbitMQ 事务channel.txSelect,然后发送消息,如果消息没有成功被 RabbitMQ 接收到,那么生产者会收到异常报错,此时就可以回滚事务channel.txRollback,然后重试发送消息;如果收到了消息,那么可以提交事务channel.txCommit。

// 开启事务 channel.txSelect(); try { // 这里发送消息 } catch (Exception e) { channel.txRollback(); // 这里再次重发这条消息 } // 提交事务 channel.txCommit();

缺点:

RabbitMQ 事务机制是同步的,你提交一个事务之后会阻塞在那儿,采用这种方式基本上吞吐量会下来,因为太耗性能。

方案2:使用confirm机制

事务机制和 confirm 机制最大的不同在于,事务机制是同步的,你提交一个事务之后会阻塞在那儿,但是 confirm 机制是异步的

在生产者开启了confirm模式之后,每次写的消息都会分配一个唯一的id,然后如果写入了rabbitmq之中,rabbitmq会给你回传一个ack消息,告诉你这个消息发送OK了;如果rabbitmq没能处理这个消息,会回调你一个nack接口,告诉你这个消息失败了,你可以进行重试。而且你可以结合这个机制知道自己在内存里维护每个消息的id,如果超过一定时间还没接收到这个消息的回调,那么你可以进行重发。

//开启confirm channel.confirm(); //发送成功回调 public void ack(String messageId){ } // 发送失败回调 public void nack(String messageId){ //重发该消息 }

2.针对RabbitMQ

主要需要应对三点:

  • 要保证rabbitMQ不丢失消息,那么就需要开启rabbitMQ的持久化机制,即把消息持久化到硬盘上,这样即使rabbitMQ挂掉在重启后仍然可以从硬盘读取消息;

  • 如果rabbitMQ单点故障怎么办,这种情况倒不会造成消息丢失,这里就要提到rabbitMQ的3种安装模式,单机模式、普通集群模式、镜像集群模式,这里要保证rabbitMQ的高可用就要配合HAPROXY做镜像集群模式;

  • 如果硬盘坏掉怎么保证消息不丢失。

(1)消息持久化

RabbitMQ 的消息默认存放在内存上面,如果不特别声明设置,消息不会持久化保存到硬盘上面的,如果节点重启或者意外crash掉,消息就会丢失。

所以就要对消息进行持久化处理。如何持久化,下面具体说明下。要想做到消息持久化,必须满足以下三个条件,缺一不可。

  • Exchange 设置持久化

  • Queue 设置持久化

  • Message持久化发送:发送消息设置发送模式deliveryMode=2,代表持久化消息

(2)设置集群镜像模式

先来介绍下RabbitMQ三种部署模式:

  • 单节点模式:最简单的情况,非集群模式,节点挂了,消息就不能用了。业务可能瘫痪,只能等待。

  • 普通模式:消息只会存在与当前节点中,并不会同步到其他节点,当前节点宕机,有影响的业务会瘫痪,只能等待节点恢复重启可用(必须持久化消息情况下)。

  • 镜像模式:消息会同步到其他节点上,可以设置同步的节点个数,但吞吐量会下降。属于RabbitMQ的HA方案

为什么设置镜像模式集群,因为队列的内容仅仅存在某一个节点上面,不会存在所有节点上面,所有节点仅仅存放消息结构和元数据。下面画了一张图介绍普通集群丢失消息情况:

如果想解决上面途中问题,保证消息不丢失,需要采用HA 镜像模式队列。

下面介绍下三种HA策略模式:

  • 同步至所有的

  • 同步最多N个机器

  • 只同步至符合指定名称的nodes

命令处理HA策略模版:

rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

1)为每个以rock.wechat开头的队列设置所有节点的镜像,并且设置为自动同步模式

rabbitmqctl set_policy ha-all "^rock.wechat" '{"ha-mode":"all","ha-sync-mode":"automatic"}' rabbitmqctl set_policy -p rock ha-all "^rock.wechat" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

2)为每个以rock.wechat.开头的队列设置两个节点的镜像,并且设置为自动同步模式

rabbitmqctl set_policy -p rock ha-exacly "^rock.wechat" \ '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

3)为每个以node.开头的队列分配指定的节点做镜像

rabbitmqctl set_policy ha-nodes "^nodes\." \ '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

但是:HA 镜像队列有一个很大的缺点就是系统的吞吐量会有所下降。

(3)消息补偿机制

为什么还要消息补偿机制呢?难道消息还会丢失,没错,系统是在一个复杂的环境,不要想的太简单了,虽然以上的三种方案,基本可以保证消息的高可用不丢失的问题。

但是作为有追求的程序员来讲,要绝对保证我的系统的稳定性,有一种危机意识。

比如:持久化的消息,保存到硬盘过程中,当前队列节点挂了,存储节点硬盘又坏了,消息丢了,怎么办?

1)生产端首先将业务数据以及消息数据入库,需要在同一个事务中,消息数据入库失败,则整体回滚。

2)根据消息表中消息状态,失败则进行消息补偿措施,重新发送消息处理。

3.针对消费者

方案一:ACK确认机制

多个消费者同时收取消息,比如消息接收到一半的时候,一个消费者死掉了(逻辑复杂时间太长,超时了或者消费被停机或者网络断开链接),如何保证消息不丢?

使用rabbitmq提供的ack机制,服务端首先关闭rabbitmq的自动ack,然后每次在确保处理完这个消息之后,在代码里手动调用ack。这样就可以避免消息还没有处理完就ack。才把消息从内存删除。

这样就解决了,即使一个消费者出了问题,但不会同步消息给服务端,会有其他的消费端去消费,保证了消息不丢的case。

总结

如果需要保证消息在整条链路中不丢失,那就需要生产端、mq自身与消费端共同去保障。

  • 生产端:对生产的消息进行状态标记,开启confirm机制,依据mq的响应来更新消息状态,使用定时任务重新投递超时的消息,多次投递失败进行报警。

  • mq自身:开启持久化,并在落盘后再进行ack。如果是镜像部署模式,需要在同步到多个副本之后再进行ack。

  • 消费端:开启手动ack模式,在业务处理完成后再进行ack,并且需要保证幂等。

通过以上的处理,理论上不存在消息丢失的情况,但是系统的吞吐量以及性能有所下降。在实际开发中,需要考虑消息丢失的影响程度,来做出对可靠性以及性能之间的权衡。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/20 13:32:06

40 个定时任务,完美诠释 RocketMQ 设计精髓!

今天来分享 RocketMQ 的定时任务。通过这些定时任务,能让我们更加理解 RocketMQ 的消息处理机制和设计理念。 从 RocketMQ 4.9.4 的源代码上看,RocketMQ 的定时任务有很多,今天主要讲解一些核心的定时任务。 1 架构回顾 首先再来回顾一下 …

作者头像 李华
网站建设 2026/9/21 16:58:40

信不信?这个 SQL 题,可以难倒大部分人

周末的时候,一个读者问了我一个很有意思的问题,是关于 MySQL 中 update 加锁的问题。 他用下面这张数据库表,做了个 MySQL 实验的时候。 发现事务 B 的 update 不会阻塞,而事务 C 的 update 会阻塞,都是对 id 10 这条…

作者头像 李华
网站建设 2026/9/20 13:32:06

40 张图,精通 Java AQS

Java中 AQS 是 AbstractQueuedSynchronizer 类,AQS 依赖 FIFO 队列来提供一个框架,这个框架用于实现锁以及锁相关的同步器,比如信号量、事件等。 在 AQS 中,主要有两部分功能,一部分是操作 state 变量,第二…

作者头像 李华
网站建设 2026/9/20 13:33:18

C++11 异步编程入门教程

1. 概述C11 引入了 <future>头文件&#xff0c;提供了一套标准化的异步编程工具&#xff0c;用于简化多线程编程和异步任务管理。这套工具主要包括 std::future、std::shared_future、std::async、std::promise、std::packaged_task等组件。2. std::futurestd::future是一…

作者头像 李华
网站建设 2026/9/20 13:33:21

【计算机毕业设计案例】基于springboot+小程序的家校通程序设计与实现教师发布作业、通知等信息,家长和学生查看信息并与教师互动(程序+文档+讲解+定制)

博主介绍&#xff1a;✌️码农一枚 &#xff0c;专注于大学生项目实战开发、讲解和毕业&#x1f6a2;文撰写修改等。全栈领域优质创作者&#xff0c;博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java、小程序技术领域和毕业项目实战 ✌️技术范围&#xff1a;&am…

作者头像 李华
网站建设 2026/9/23 2:09:19

小程序毕设选题推荐:基于springboot+小程序的高校校园信息交流平台小程序基于微信小程序的校园信息交流平台【附源码、mysql、文档、调试+代码讲解+全bao等】

博主介绍&#xff1a;✌️码农一枚 &#xff0c;专注于大学生项目实战开发、讲解和毕业&#x1f6a2;文撰写修改等。全栈领域优质创作者&#xff0c;博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java、小程序技术领域和毕业项目实战 ✌️技术范围&#xff1a;&am…

作者头像 李华