【RabbitMQ | 第六篇】消息重复消费问题及解决方案

这篇具有很好参考价值的文章主要介绍了【RabbitMQ | 第六篇】消息重复消费问题及解决方案。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

rabbitmq怎么解决重复消费,# 【RabbitMQ】,rabbitmq,分布式

6.消息重复消费问题

6.1问题介绍

什么是消息重复消费?首先我们来看一下消息的传输流程。消息生产者–>MQ–>消息消费者;消息生产者发送消息到MQ服务器,MQ服务器存储消息,消息消费者监听MQ的消息,发现有消息就消费消息。

所以消息重复也就出现在 两个阶段

1 :生产者多发送了消息给MQ;

2 :MQ的一条消息被消费者消费了多次

具体场景如下:

  1. 生产者发送消息给MQ在MQ确认的时候出现了网络波动,生产者没有收到确认,这时候生产者就会重新发送这条消息,导致MQ会接收到重复消息。
  2. 消费者消费成功后,给MQ确认的时候出现了网络波动,MQ没有接收到确认,为了保证消息不丢失,MQ就会继续给消费者投递之前的消息。这时候消费者就接收到了两条一样的消息。由于重复消息是由于网络原因造成的,无法避免。

6.2解决思路

  1. 发送消息时让每个消息携带一个全局的唯一ID
  2. 在消费消息时先判断消息是否已经被消费过,保证消息消费逻辑的幂等性。具体消费过程为:
    • 消费者获取到消息后先根据id去查询redis/db是否存在该消息
    • 如果不存在,则正常消费,消费完毕后写入redis/db
    • 如果存在,则证明消息被消费过,直接丢弃

6.3将该消息存储到Redis

6.3.1将id存入string(单消费者场景)

(1)实现思路
  • 将id号存入value中,并且value类型为string
  • 即以队列名称为key,以消息id为值
  • 每次消息过来都覆盖之前的消息
    @RabbitListener(queues = "queueName4")//发送的队列名称     @RabbitListener注解到类和方法都可以
    @RabbitHandler
    public void receiveMessage1(Message message) throws UnsupportedEncodingException {
        //获取唯一id
        String messageId = message.getMessageProperties().getMessageId();
        String msg = new String(message.getBody(),"utf-8");
 		
        //获取redis中该队列名称对应的value值
        String messageRedisValue = redisUtil.get("queueName4","");
        
        //检验唯一id是否存在
        if (messageRedisValue.equals(messageId)) {
            //存在
            return;
        }
        System.out.println("消息:"+msg+", id:"+messageId);
 		
        //以队列为key,id为value
        redisUtil.set("queueName4",messageId);
    }
(2)问题
  1. 并发冲突:如果多个消费者同时操作 Redis 中的已消费消息列表,由于 Redis 是单线程处理命令,可能会出现并发冲突导致数据不一致或丢失问题。特别是在高并发情况下,使用字符串类型的 ID 可能会增加并发冲突的风险
  2. 内存占用:字符串类型的 ID 在内存中占用空间相对较大,尤其是对于大量消息的情况下,会增加 Redis 的内存占用。
  3. 比较效率:字符串类型的 ID 比较起来相对复杂,需要进行字符串比较操作。

6.3.2将id存入list中(多消费场景)

(1)实现思路
  • 以该队列名称为key,id为value
  • 适合多消费场景的原因:
    • 顺序性:List 是一个有序集合,可以按照消息的顺序存储消息 ID。在多消费者场景下,保持消息的顺序通常是很重要的,以确保消息按照正确的顺序被消费。
    • 原子性操作:Redis 的 List 提供了多个原子性操作,比如从列表两端推入/弹出元素,这些操作可以确保多个消费者同时访问列表时不会出现数据竞争和并发问题。
    • 支持阻塞操作:List 提供了阻塞式的弹出操作(如 BLPOP、BRPOP),可以在没有消息时阻塞等待新消息的到来,这对于实现消费者轮询机制非常有用。
@RabbitListener(queues = "queueName4")//发送的队列名称     @RabbitListener注解到类和方法都可以
@RabbitHandler
public void receiveMessage2(Message message) throws UnsupportedEncodingException {
    String messageId = message.getMessageProperties().getMessageId();
    String msg = new String(message.getBody(),"utf-8");
 	
    //获取
    List<String> messageRedisValue = redisUtil.lrange("queueName4");
    if (messageRedisValue.contains(messageId)) {
        return;
    }
    System.out.println("消息:"+msg+", id:"+messageId);
 
    redisUtil.lpush("queueName4",messageId);//存入list
}

6.3.3将id以key增量存入string中并设置过期时间

(1)实现思路

消息id为key消息内容为value存入string中,设置过期时间( 可承受的redis服务器异常时间,比如设置过期时间为10分钟,如果redis服务器断了20分钟,那么未消费的数据都会丢了)

    @RabbitListener(queues = "queueName4")//发送的队列名称     @RabbitListener注解到类和方法都可以
    @RabbitHandler
    public void receiveMessage2(Message message) throws UnsupportedEncodingException {
        String messageId = message.getMessageProperties().getMessageId();
        String msg = new String(message.getBody(),"utf-8");
 
        String messageRedisValue = redisUtil.get(messageId,"");
        if (msg.equals(messageRedisValue)) {
            return;
        }
        System.out.println("消息:"+msg+", id:"+messageId);
 		
        //以id为key,消息内容为value,过期时间10分钟
        redisUtil.set(messageId,msg,10L);
    }

6.4总结

该篇文章介绍了消息重复消费问题及解决方案,问题可能产生的两个阶段(生产消息多发、消费者重复消息);解决方案:将消息发送时携带一个唯一id,消费方拿到消息时先去reids/db中有没有该数据,若没有则可以消费,否则不可以消费;并介绍了基于Redsi解决消息重复消费问题,①以队列名称为key,消息id为value,且value为string类型(适合只有一个消费方)②以队列名称为key,消息id为value,且value为list类型(适合有多个消费方场景)③以消息id为key,内容为value,并设置过期时间

rabbitmq怎么解决重复消费,# 【RabbitMQ】,rabbitmq,分布式文章来源地址https://www.toymoban.com/news/detail-858423.html

到了这里,关于【RabbitMQ | 第六篇】消息重复消费问题及解决方案的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • mq常见问题:消息丢失、消息重复消费、消息保证顺序

    mq常见问题:消息丢失、消息重复消费、消息保证顺序 消息丢失问题 拿rabbitmq举例来说,出现消息丢失的场景如下图 从图中可以看到一共有以下三种可能出现消息丢失的情况: 1 生产者丢消息 生产者在将数据发送到MQ的时候,可能由于网络等原因造成消息投递失败 2MQ自身丢

    2024年02月09日
    浏览(55)
  • MQ消息队列详解以及MQ重复消费问题

    https://blog.csdn.net/qq_44240587/article/details/104630567 核心的就是:解耦、异步、削锋 现有ABCDE五个系统,最初的时候BCD三个系统都要调用A系统的接口获取数据,一切都很正常,但是突然,D系统说:我不要了,你不用给我传数据了,A系统无奈,只能修改代码,将调用D系统的代码删除

    2024年04月13日
    浏览(52)
  • RabbitMQ消费端消费能力不足,消息消费慢问题解决思路

    思路:(易-难) 1.新增服务器,增加消费端数量。 2.服务器端开启多线程消费消息(默认单线程监听队列),需要注意多线程可能带来的并发问题。 3.优化消费端处理逻辑,提升消息消费速率即系统优化。 背景: 本系统为小型支付系统,商户618活动期间,用户购买商品付款

    2024年02月13日
    浏览(84)
  • JAVA面试题分享一百六十二:Kafka消息重复消费问题?

    消息重复消费的根本原因都在于:已经消费了数据,但是offset没有成功提交。 其中很大一部分原因在于发生了再均衡。 1)消费者宕机、重启等。导致消息已经消费但是没有提交offset。 2)消费者使用自动提交offset,但当还没有提交的时候,有新的消费者加入或者移除,发生

    2024年02月03日
    浏览(41)
  • 记一次rabbitmq消息发送成功,消费丢失问题

    测试数据归档,偶现数据未归档 idea线上调试,log日志,数据库消息发送记录,代码分块重复执行看哪块出的问题,结果均无问题,最后使用rabbitmq trace日志,发现问题所在 1. 什么是Trace Trace 是Rabbitmq用于记录每一次发送的消息,方便使用Rabbitmq的开发者调试、排错。可通过插

    2024年02月16日
    浏览(47)
  • RabbitMq 的消息可靠性问题(二)---MQ的消息丢失和consumer消费问题

    RabbitMq 消息可靠性问题(一) — publisher发送时丢失 前面我们从publisher的方向出发解决了发送时丢失的问题,那么我们在发送消息到exchange, 再由exchange转存到queue的过程中。如果MQ宕机了,那么我们的消息是如何确保可靠性的呢?当消息由队列发到对应的消费者处理时,consumer 接

    2024年02月11日
    浏览(42)
  • Kafka如何保证消息的消费顺序【全局有序、局部有序】、Kafka如何保证消息不被重复消费、Kafka为什么这么快?【重点】、Kafka常见问题汇总【史上最全】

    目录 Kafka消息生产 一个Topic对应一个Partition 一个Topic对应多个Partition Kafka消息的顺序性保证(Producer、Consumer) 全局有序 局部有序  max.in.flight.requests.per.connection参数详解 Kafka的多副本机制 Kafka的follower从leader同步数据的流程 Kafka的follower为什么不能用于消息消费 Kafka的多分区

    2024年04月11日
    浏览(51)
  • RabbitMQ中channel与try()结合导致的消息不消费或消息无法接收的问题分析与定位

    RabbitMQ中channel与try()结合导致的消息不消费或消息无法接收的问题分析与定位 使用 rabbitMQ 测试 topic 交换机的案例 关键信息 RabbitMQ、 try、消费者无法接收消息 想要直接参考解决方案,看最后一部分 想要看问题原因,看第三部分 想看如何分析,顺序浏览 个人认为,交换机相

    2024年02月13日
    浏览(34)
  • RabbitMQ - 消息堆积问题的最佳解决方案?惰性队列

    目录 一、惰性队列 1.1、消息堆积问题 1.2、消息堆积问题的解决方法 从消费者的角度: 从队列的角度: 1.3、引入惰性队列 1.3.1、什么是惰性队列 1.3.2、惰性队列的使用 1.3.3、效果演示 当生产者发送消息的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,直到

    2024年02月05日
    浏览(37)
  • rabbitmq topic模式设置#通配符情况下 消费者队列未接收消息问题排查解决

    生产者配置 Exchange:topic_exchange_shcool Routing key:topic.shcool.# 消费者代码配置 Exchange:topic_exchange_shcool Routing key:topic.shcool.user 其实以上代码看着没有问题,意思是代码生成一个队列,并把【topic.shcool.user】队列和生产者的【topic_exchange_shcool】exchange绑定,但是生产者发送消息是

    2024年02月11日
    浏览(44)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包