RabbitMQ避免重复消费

2023-12-05

在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

事务性消费
消费状态追踪
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

RabbitMQ避免重复消费 的相关文章

  • Symfony Messenger 和 Mailer:如何添加绑定密钥?

    我有一个正在运行的 Symfony 4 4 项目 其中包含Messenger和rabbitMQ 我有一个带有 2 个队列的异步传输 transports https symfony com doc current messenger htm
  • 每次发送消息时是否需要重建RabbitMQ连接

    我有一个 Spring 3 应用程序 它通过非 RabbitMQ 接收器接收消息 处理它们并通过 RabbitMQ 转发 每次发送消息时都会建立一个新的 RabbitMQ 连接 这似乎有点浪费 我只是想知道这是否真的有必要 或者是否有原因导
  • RabbitMQ 中的并发

    经过一周的编码和搜索论坛后 似乎是时候问 我有一个 C 应用程序 它使用 EventingBasicConsumer 处理 RabbitMQ 发送的消息 我想同时处理多个消息 因此我在同一连接上实例化了几个通道 本例中为 8 个 每个通道都
  • 如何优雅地结束 spring @Schedule 任务?

    我正在尝试让 Spring Boot 服务优雅地结束 它有一个方法 Scheduled注解 该服务使用 spring data 作为数据库 使用 spring cloud stream 作为 RabbitMQ 在计划的方法结束之前 数据库和
  • Camel 中的无限循环 - Rabbitmq

    我有一个小型服务器路由 它从queue in 获取消息并放入queue out 当我输入一条消息时queue in 服务器无限循环运行 我想知道我在配置方面缺少什么 这看起来是一条非常简单的路线 小服务器路由
  • 在使用 FromEventPattern 订阅之前捕获事件

    我正在使用 Rx 框架编写消息监听器 我面临的问题是 我正在使用的库使用一个消费者 每当消息到达时就会发布事件 我已经设法通过以下方式消费传入的消息Observable FromEventPattern但我对服务器中已有的消息有疑问 目前我
  • amqp 与 amqplib - 哪个 Node.js amqp 客户端库更好?

    这些 amqp 客户端库之间有什么区别 哪一款最值得推荐 主要区别是什么 我会推荐amqp node https github com squaremo amqp node and bramqp https github com bakke
  • 在 Docker 中使用 RabbitMQ 删除容器

    我尝试使用RabbitMQ启动Docker容器 结果下载了镜像 但容器没有启动 我在日志中收到以下消息 error RABBITMQ DEFAULT PASS is set but deprecated error RABBITMQ DEF
  • 在点网核心应用程序中使用 RabbitMQ 跳过 MassTransit 中的队列

    我有三个项目 一个是Dot net core MVC 两个是API项目 MVC 正在调用一个 API 来获取用户详细信息 当询问用户详细信息时 我通过 MassTransit 向队列发送消息 我看到跳过队列 第三个项目中有消费者 即API项
  • 如何触发 IModel.BasicAcks?

    我第一次使用 RabbitMQ 的 NET API 我想出了一个对我来说似乎合理的用例 我想创建发布消息并在消息被确认后执行某些操作的发布者 IModel BasicAcks 事件似乎是了解这一点的好方法 所以 我给出版商写了一封信 pri
  • 使用Camel的spring-rabbitmq组件时如何自动声明交换?

    我正在尝试从 Camel 3 x 迁移到 Camel 4 x 版本 因此我需要从rabbitmq替换组件spring rabbitmq With rabbitmq我正在使用的组件declare https camel apache org
  • 在 RabbitMQ 监听器中隐藏运行时异常

    在某些故意发生的情况下 我使用了一些异常来拒绝消息 但在控制台中显示了乍一看似乎不太正常的异常 如何在登录控制台 文件时隐藏该特定异常 我正在使用 spring boot 和默认记录器 public static class Undispa
  • 无法在Windows上启用rabbitmq管理插件

    所以 这就是我所做的 在我的 Windows x64 位机器上安装了 Erlang 安装 RabbitMQ 启动 RabbitMQ 服务 这一步我没有任何错误 但是 当我尝试启用rabbitmq management时 我在控制台中收到一些
  • Celery 任务状态取决于 CELERY_TASK_RESULT_EXPIRES

    据我所知 任务状态完全取决于 CELERY TASK RESULT EXPIRES 设置的值 如果我在任务完成执行后检查此间隔内的任务状态 则返回的状态为 AsyncResult task id state 是正确的 如果没有 状态将不会更
  • springrabbitmq:无法将id设置为属性?

    我有一个属性文件 其中包含队列 其值为queue name 如果我在其他请使用该属性 那么它可以工作 但如果我在 id 上使用它 那么它会失败
  • 在 Celery 工作线程中捕获 Heroku SIGTERM 以优雅地关闭工作线程

    我对此进行了大量研究 令我惊讶的是我还没有在任何地方找到一个好的答案 我正在 Heroku 上运行一个大型应用程序 并且我有某些运行很长时间处理的 celery 任务 并在任务结束时保存结果 每次我在 Heroku 上重新部署时 它都会发送
  • RabbitMQ Java 客户端自动重新连接

    当我的应用程序失去与 RabbitMQ 的连接时 我将其连接工厂设置为自动尝试并重新连接 ConnectionFactory factory new ConnectionFactory factory setUsername usernam
  • MassTransit 生成我想忽略的_skipped 队列

    任何人都可以猜出问题是什么 因为我不知道如何解决这个问题 大众运输产生 skipped队列 我不知道为什么它会生成这些队列 它是在执行发布请求响应时生成的 请求客户端是使用 MassTransit RequestClientExtensio
  • AMQPRuntimeException:读取数据时出错。收到 0 而不是预期的 7 字节

    它曾经有效 但现在不再有效了 我正在使用 php amqplib 和 RabbitMQ 当我尝试创建新的 AMQP 连接时 connection new AMQPConnection localhost 5672 username pass
  • 基于多线程的 RabbitMQ 消费者

    我们有一个 Windows 服务 它监听单个 RabbitMQ 队列并处理消息 我们希望扩展相同的 Windows 服务 以便它可以监听 RabbitMQ 的多个队列并处理消息 不确定使用多线程是否可以实现这一点 因为每个线程都必须侦听 阻

随机推荐