RabbitMQ避免重复消费

在Java中,可以使用消息队列来实现消息的异步处理,其中常用的消息队列有 RabbitMQ、ActiveMQ、Kafka 等。

什么是幂等性?

幂等性是指无论操作执行多少次,都是得到相同的结果,而不会产生其他副作用。

在rabbitMQ中

什么是消息重复消费?

同一条消息在MQ中被消费多次

出现重复消费的原因:

生产者发送一条消息到rabbitMQ,但rabbitMQ尚未收到消费者的确认,会认为消息消费未被消费而重新发送。

网络不稳定、消费者故障、网络分区、消息重复传递策略、消费者超时设置不当

为什么需要避免重复消费?

业务错误:我本来写的业务逻辑就是只要执行一次

数据重复:数据插入重复,破坏数据唯一性

资源浪费:占用系统资源,降低系统性能

如何避免消息重复消费?
消息去重

通过记录已经消费过的消息,在消息到达时检查它是否已经在记录中存在,从而避免重复处理。

  if (!processedMessages.contains(message)) {
   
                    processMessage(message);
                    processedMessages.add(message);
                }

消息幂等性
分布式锁(消息幂等性)

使用UUID生成唯一Id ,作为messageId

使用了唯一的消息ID来确保同一条消息只会被处理一次。

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                    .messageId(UUID.randomUUID().toString()) // 唯一标识
                    .build();

        if (!isMessageProcessed(messageId)) {
   
            processMessage(message);
            saveProcessedMessage(messageId);
        }

消费者先查询该消息是否已经被处理过,如果没有被处理过,则调用processMessage()方法处理该消息,并使用 saveProcessedMessage()方法保存已经处理过的消息。

            //手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false

在处理完消息后,还需要调用channel.basicAck(envelope.getDeliveryTag(), false)方法确认消息已经被消费。这是因为RabbitMQ是一个消息的投递机制,只有在消费者确认了消息已经被处理后,才会从消息队列中删除该消息。

使用redis实现避免重复消费

生产者

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                    .deliveryMode(1) // 指定消息是否需要持久化 1-需要 2-不需要
                    .messageId(UUID.randomUUID().toString()) // 唯一标识
                    .build();

消费者

 String result = jedis.set(messageId, "0", "NX", "EX", 10);
  
 if (result != null && result.equalsIgnoreCase("OK")){
   
                    System.out.println("接收到消息:"+ new String(body,"UTF-8"));

                    //消费成功 set messageId - 1
                    jedis.set(messageId,"1");
                    channel.basicAck(envelope.getDeliveryTag(),false);

                }else {
   
                    //如果1中的setnx失败,获取key对应的value,如果是1,设置ack 如果是0 return

                    String s = jedis.get(messageId);
                    if ("1".equalsIgnoreCase(s)){
   
                        //消费完了
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }

spring-boot

如果存在,设置value为1;如果value是1,ack

事务性消费
消费状态追踪

相关推荐

  1. RabbitMQ避免重复消费

    2023-12-05 15:27:43       58 阅读
  2. Rabbitmq消息重复消费

    2023-12-05 15:27:43       19 阅读
  3. react native hooks 如何避免重复请求

    2023-12-05 15:27:43       16 阅读

最近更新

  1. TCP协议是安全的吗?

    2023-12-05 15:27:43       16 阅读
  2. 阿里云服务器执行yum,一直下载docker-ce-stable失败

    2023-12-05 15:27:43       16 阅读
  3. 【Python教程】压缩PDF文件大小

    2023-12-05 15:27:43       15 阅读
  4. 通过文章id递归查询所有评论(xml)

    2023-12-05 15:27:43       18 阅读

热门阅读

  1. 简谈PostgreSQL的wal_level=logic

    2023-12-05 15:27:43       31 阅读
  2. FlinkSql-Temporal Joins-Lookup Join

    2023-12-05 15:27:43       47 阅读
  3. postgresql树状结构查询示例

    2023-12-05 15:27:43       41 阅读
  4. 高级软件工程15本书籍

    2023-12-05 15:27:43       38 阅读
  5. flask 请求勾子实现 request_auth认证

    2023-12-05 15:27:43       37 阅读
  6. 第九章 Flask

    2023-12-05 15:27:43       39 阅读
  7. 关于运维工作和Linux的理解

    2023-12-05 15:27:43       35 阅读
  8. html中css样式匹配替换

    2023-12-05 15:27:43       37 阅读
  9. excel 底部不显示sheet 名称,怎么设置可以现实

    2023-12-05 15:27:43       60 阅读