聊聊MQ,如何避免消息丢失?如何避免重复消费?

柳土獐
• 阅读 1796

前言

我在工作中,使用到消息中间件MQ的业务还是挺多的,我从事在一家交通行业的公司,业务中经常会涉及处理一些违法数据的场景,项目中经常会使用到RabbitMQ,今天想跟大家聊聊怎样避免消息丢失和重复消费是问题。

聊聊MQ,如何避免消息丢失?如何避免重复消费?

在我们发生消息的时候,如果MQ服务器突然宕机了会出现什么情况?是不是我们发过去的消息全部没有了吗?

是的,一般MQ中间件为了提高系统的吞吐量会把消息保存在内存中,如果不作其他处理,MQ服务器一旦宕机,消息将全部丢失。这个是业务不允许的,造成很大的影响。

在遇到这种问题的时候,一般有以下几种处理方式。

如何避免消息丢失

持久化

RabbitMQ中发消息的时候会有个durable参数可以设置,设置为true,就会持久化。

聊聊MQ,如何避免消息丢失?如何避免重复消费?

这样的话MQ服务器即使宕机,重启后磁盘文件中有消息的存储,这样就不会丢失了吧。是的这样就一定概率的保障了消息不丢失。

但还会有个场景,就是消息刚刚保存到MQ内存中,但还没有来得及更新到磁盘文件中,突然宕机了。这个场景在持续的大量消息投递的过程中,会很常见。

那怎么办?我们如何作才能保障一定会持久化到磁盘上面呢?

confirm机制

RabbitMQ利用confirm机制来通知我们是否持久化成功?

聊聊MQ,如何避免消息丢失?如何避免重复消费?

confirm机制的原理:

(1)消息生产者把消息发送给MQ,如果接收成功,MQ会返回一个ack消息给生产者;

(2)如果消息接收不成功,MQ会返回一个nack消息给生产者;

这样是不是就可以保障100%消息不丢失了呢?

如果我们生产者每发一条消息,都要MQ持久化到磁盘中,然后再发起ack或nack的回调。这样的话是不是我们MQ的吞吐量很不高,因为每次都要把消息持久化到磁盘中。写入磁盘这个动作是很慢的。这个在高并发场景下是不能够接受的,吞吐量太低了。

所以MQ持久化磁盘真实的实现,是通过异步调用处理的,他是有一定的机制,如:等到有几千条消息的时候,会一次性的刷盘到磁盘上面。而不是每来一条消息,就刷盘一次。

所以comfirm机制其实是一个异步监听的机制,是为了保证系统的高吞吐量,这样就导致了还是不能够100%保障消息不丢失,因为即使加上了confirm机制,消息在MQ内存中还没有刷盘到磁盘就宕机了,还是没法处理。

数据库事务机制

生产者在投递消息之前,可以在本地数据库建一张消息表,先把消息持久化到Redis或DB中,这样就可以利用本地数据库的事务机制。事务提交成功后,将消息表中的消息转移到消息队列中。

confirm机制监听消息是否发送成功?如ack成功消息,删除DB中此消息。

如果nack不成功的消息,这个可以根据自身的业务选择是否重发此消息。也可以删除此消息,由自己的业务决定。

如何避免消息重复消费

幂等性也就是相同条件下对一个业务的操作,不管操作多少次,结果都是一样。

重复消费的问题?

导致重复消费的原因可能出现在生产者,也可能出现在 MQ 或 消费者。

这里说的重复消费问题是指同一个数据被执行了两次,不单单指 MQ 中一条消息被消费了两次,也可能是 MQ 中存在两条一模一样的消费。

  • 生产者:生产者可能会重复推送一条数据到 MQ 中,为什么会出现这种情况呢?也许是一个 Controller 接口被重复调用了 2 次,没有做接口幂等性导致的;也可能是推送消息到 MQ 时响应比较慢,生产者的重试机制导致再次推送了一次消息。
  • MQ:在消费者消费完一条数据响应 ack 信号消费成功时,MQ 突然挂了,导致 MQ 以为消费者还未消费该条数据,MQ 恢复后再次推送了该条消息,导致了重复消费。
  • 消费者:消费者已经消费完了一条消息,正准备但是还未给 MQ 发送 ack 信号时,此时消费者挂了,服务重启后 MQ 以为消费者还没有消费该消息,再次推送了该条消息。

如何保证幂等性?

消费者怎么解决重复消费问题呢?这里提供两种方法:

  • 状态判断法:消费者消费数据后把消费数据记录在 redis 中,下次消费时先到 redis 中查看是否存在该消息,存在则表示消息已经消费过,直接丢弃消息。
  • 业务判断法:通常数据消费后都需要插入到数据库中,使用数据库的唯一性约束防止重复消费。每次消费直接尝试插入数据,如果提示唯一性字段重复,则直接丢失消息。一般都是通过这个业务判断的方法就可以简单高效地避免消息的重复处理了。
点赞
收藏
评论区
推荐文章
Oracle 分组与拼接字符串同时使用
SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(
Wesley13 Wesley13
4年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Centos7安装RabbitMQ详细教程 - 附带软件基本解释 - CSDN博客
MQ引言什么是MQMQ:messageQueue翻译为消息队列,通过典型的生产者和消费者模型不断向消息队列中生产消息,消费者不断从队列中获取消息。因为消息的生产和消费都是一部的,而且只关心消息的发送和接收,没有业务逻辑的侵入,轻松的实现了系统之间的解耦。别名是消息中间件,通过利用高效的消息传递机制进行与平台无关的数据交流,并基于数据通信来进行分布式系
Stella981 Stella981
4年前
RabbitMq学习笔记——概念
1、RabbitMQ简介  MQ全称为MessageQueue(消息队列),是一种“应用程序”<—“应用程序”的通信方法。MQ是一个典型的“消费”<—“生产者”模型的代表,生成者往消息队列中写入消息,消费者从消息队列中读取消息。2、MQ的应用场景  对于一个大型的软件系统来说,它会有很多的组件或者说模块或者说子系统或者
Stella981 Stella981
4年前
RabbitMQ如何保证队列里的消息99.99%被消费?
1\.本篇概要其实,还有1种场景需要考虑:当消费者接收到消息后,还没处理完业务逻辑,消费者挂掉了,那消息也算丢失了?,比如用户下单,订单中心发送了1个消息到RabbitMQ里的队列,积分中心收到这个消息,准备给这个下单的用户增加20积分,但积分还没增加成功呢,积分中心自己挂掉了,导致数据出现问题。那么如何解
Wesley13 Wesley13
4年前
mq要如何处理消息丢失、重复消费?
如果要你实现一个支付宝向余额宝转账的功能,比如:账户a从支付宝转出5000余额宝转入5000,该怎么做呢?可能有些人会说,这还不简单,直接上图!(https://oscimg.oschina.net/oscnet/7d47d132357446109035157b25361ec7.png)支付宝先给账户a减5000,调用余额
Stella981 Stella981
4年前
RabbitMQ 如何保证消息的可靠性
一条消费成功被消费经历了生产者MQ消费者,因此在这三个步骤中都有可能造成消息丢失。一消息生产者没有把消息成功发送到MQ1.1事务机制AMQP协议提供了事务机制,在投递消息时开启事务支持,如果消息投递失败,则回滚事务。自定义事务管理器@Configuration
Wesley13 Wesley13
4年前
MQ 消息队列
1、场景作用削峰填谷,异步解耦。2、如何保证消息不被重复消费呢?这个问题可以换个思路,保证消息重复消费,其实是保证程序的幂等性。无论消息如何重复,程序运行的结果是一致的。比如消费消息后做数据库插入操作,为了防止消息重复消费,可以在插入前先查询一下有没有对应的数据。3、怎么保证从消息队列里拿到的数据按顺序执
MQ消息乱序问题解析与实战解决方案
作者:京东物流刘浩1.背景在分布式系统中,消息队列(MQ)是实现系统解耦、异步通信的重要工具。然而,MQ消费时出现的消息乱序问题,经常会对业务逻辑的正确执行和系统稳定性产生不良影响。本文将详细探讨MQ消息乱序问题的根源,并提供一系列在实际应用中可行的解决方