草庐IT

RabbitMQ消息确认机制和消息重发机制

Sora33 2023-05-03 原文

一.机制

首先我们要知道一条消息的传递过程。

生产者 -> 交换机 ->  队列

我们的生产者生产消息,生产完成的消息发送到交换机,由交换机去把这个消息转发到对应的队列上。这其中我们可能在生产者 -> 交换机丢失消息,也可能在 交换机 -> 队列上丢失消息。因此我们需要引入2个概念。

1: 生产者到交换机的可靠保证 (confirmCallback ) 确认回调机制

2: 交换机到队列的保证 (returnCallback ) 返回回调机制

二. 保证生产者到交换机的可靠传递

因为我们的消息都要经过路由,然后去对应的队列,所以第一条线路至关重要。我们使用confirm机制。这个confirm机制是一个异步的,也就是说我们发送一条消息之后可以继续发送下一条消息。比自带的事务好很多。

使用confirm机制首先需要在配置文件中开启confirm机制

  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /
    username: admin
    password: password
    # 开启生产者消息确认
    publisher-confirm-type: correlated

生产者代码

@GetMapping("/send/{tel}")
    public Result send(@PathVariable("tel") String tel) {
        // 开启生产者回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if (b) {
                    log.info("消息发送到交换机成功");
                } else {
                    log.error("消息发送到交换机失败,失败信息[{}]",s);
                }
            }
        });

        // 发送消息
        rabbitTemplate.convertAndSend(RabbitConstant.DIRECT_EXCHANGE,"sms",tel);

        return null;
    }

这样我们的消息如果发送到交换机,就会执行[消息发送到交换机成功](签收成功是后面消费者里的消息签收机制,现在不用在意)

 现在我们测试一下,我在交换机名字后面加上一个字符串,现在这个交换机是不存在的。看看会发生什么

rabbitTemplate.convertAndSend(RabbitConstant.DIRECT_EXCHANGE+"sora33","sms",tel);

 发送到交换机失败了,执行了我们失败里的回调。这里我们就可以看出这个confirm机制的作用了。它是用来确保确保我们的消息是否到达了交换机。到达了执行ack,没有到达执行nack。我们可以在发送失败的方法里加入自己的逻辑。比如加入到发送失败的表中,或者尝试重新发送...

消息的发送确认机制讲完之后。接下来我们来看一下交换机到队列要如何保证消息的可靠性。

三.保证交换机到队列的可靠传递

使用ReturnCallback机制来保证。假设我现在有一个路由模式的交换机。绑定了一个队列,叫send_sms。对应的路由键是sms。如果我给这个交换机发送一条消息。路由键指定smssss。肯定是找不到对应的队列。那么这个时候就会触发ReturnCallback。

setMandatory是用来设置如果没有找到队列,是丢弃还是执行returnedMessage里的方法。false丢弃。

要使用ReturnCallback,我们同样需要在设置中打开配置,很简单。只需要在yml里的mq下面跟一条配置就行了。打开return回调机制

publisher-returns:true

 加入下面的回调属性设置。可以和消息确认机制一起使用。2者互不影响,直接写上去就行。

 // 队列收到消息确认机制
 rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
 @Override
 public void returnedMessage(Message message, int i, String context, String exchange, 
 String routeKey) {
    log.error("消息[{}]未到达队列[{}],使用的路由键[{}]",message,exchange,routeKey);
   }
 });
 // true -> 消息未到队列中触发MessageReturn false -> 消息未到队列直接丢弃该消息
 rabbitTemplate.setMandatory(true);

 现在我给一个不存在的路由key发送。交换机肯定是找不到对应的队列的 我们的交换机目前只绑定了路由为sms的一个队列 

rabbitTemplate.convertAndSend(RabbitConstant.DIRECT_EXCHANGE,"smssss",tel);

 可以看到,虽然消息进入了交换机,但是找不到对应的队列,执行ReturnCallback回调函数

生产者方面的一些机制讲完之后。接下来我们来看消费者中的消息签收机制以及如何重新发送失败的消息。

因为rabbitMQ默认是签收消息的。我们先把签收模式设置为手动签收 顺便配置一下我们的重发配置

# 消费端设置手动签收
    listener:
      direct:
        acknowledge-mode: manual
      simple:
        acknowledge-mode: manual
        retry:
          # 开启消息重发机制
          enabled: true
          # 重试次数3
          max-attempts: 3

生产者代码 生产者的逻辑很简单。肯定会抛异常。因为我手动设置了一个被除数异常。进入到catch块中。我做了一个存入redis 的操作,将这个消息的标签值作为键。值设置为1.作为重试次数。存入之后mq会自动进行一个重发。当判断重试次数达到3次。直接拒绝签收。并将该消息存到数据库中的重试表。进行一个人工操作...

    int a = 0;

    @RabbitListener(queues = {RabbitConstant.SEND_SMS})
    public void smsQueue(String tel, Message message, Channel channel) throws IOException {
        try {
            int c = 1/a;
            // 签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            log.info("签收成功[{}]",tel);
        } catch (Exception e) {
            // 获取redis重试次数
            Integer value = (Integer)redisUtil.get(message.getMessageProperties().getDeliveryTag() + "");
            if (value == null) {
                // 存入redis
                redisUtil.set(message.getMessageProperties().getDeliveryTag()+"", 1);
            } else if (value.intValue() == 2) { // 如果第三次还是有异常,那么第三次的次数value值还是2 所以加入重试表
                // logic // 加入重试表
                log.error("消息[{}]消费失败...传递参数[{}]", message, tel);
                log.warn("已加入重试表...");
                // 签收失败并不重试
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                return;
            } else {
                redisUtil.set(message.getMessageProperties().getDeliveryTag() + "", ++value);
            }
            log.info("签收失败[{}]",tel);
            throw new RuntimeException("签收异常");
        }
    }

 当我们给int c = 1/a 改为 c = 1/a++

这个时候第一次会进入catch块。第二次因为a自增。所以不会抛出异常,签收成功

 四.总结

        RabbitMQ在我们工作中是常用的一个中间件,必须要对齐了如指。既然是中间件,那么势必会有消息丢失产生,还要保证消息的幂等性。本文章是一个进阶文章。不懂得小伙伴可以去看看我的前两篇。

springBoot使用RabbitMq6大模式详解_Sora33的博客-CSDN博客_springboot 使用rabbitmq最简单最全的rabbitmq上手教程https://blog.csdn.net/qq_40179653/article/details/125589283

rabbitMq延迟队列的使用_Sora33的博客-CSDN博客主要介绍rabbitmq延迟队列的使用https://blog.csdn.net/qq_40179653/article/details/125618655?spm=1001.2014.3001.5502

 更多知识请移步个人博客:33sora.com​​​​​​​

有关RabbitMQ消息确认机制和消息重发机制的更多相关文章

  1. ruby-on-rails - 如何在 Rails View 上显示错误消息? - 2

    我是rails的新手,想在form字段上应用验证。myviewsnew.html.erb.....模拟.rbclassSimulation{:in=>1..25,:message=>'Therowmustbebetween1and25'}end模拟Controller.rbclassSimulationsController我想检查模型类中row字段的整数范围,如果不在范围内则返回错误信息。我可以检查上面代码的范围,但无法返回错误消息提前致谢 最佳答案 关键是您使用的是模型表单,一种显示ActiveRecord模型实例属性的表单。c

  2. ruby - 使用 Ruby 通过 Outlook 发送消息的最简单方法是什么? - 2

    我的工作要求我为某些测试自动生成电子邮件。我一直在四处寻找,但未能找到可以快速实现的合理解决方案。它需要在outlook而不是其他邮件服务器中,因为我们有一些奇怪的身份验证规则,我们需要保存草稿而不是仅仅发送邮件的选项。显然win32ole可以做到这一点,但我找不到任何相当简单的例子。 最佳答案 假设存储了Outlook凭据并且您设置为自动登录到Outlook,WIN32OLE可以很好地完成此操作:require'win32ole'outlook=WIN32OLE.new('Outlook.Application')message=

  3. Ruby - 如何将消息长度表示为 2 个二进制字节 - 2

    我正在使用Ruby,我正在与一个网络端点通信,该端点在发送消息本身之前需要格式化“header”。header中的第一个字段必须是消息长度,它被定义为网络字节顺序中的2二进制字节消息长度。比如我的消息长度是1024。如何将1024表示为二进制双字节? 最佳答案 Ruby(以及Perl和Python等)中字节整理的标准工具是pack和unpack。ruby的packisinArray.您的长度应该是两个字节长,并且按网络字节顺序排列,这听起来像是n格式说明符的工作:n|Integer|16-bitunsigned,network(bi

  4. ruby-on-rails - 在 Flash 警报 Rails 3 中显示错误消息 - 2

    如果我在模型中设置验证消息validates:name,:presence=>{:message=>'Thenamecantbeblank.'}我如何让该消息显示在闪光警报中,这是我迄今为止尝试过的方法defcreate@message=Message.new(params[:message])if@message.valid?ContactMailer.send_mail(@message).deliverredirect_to(root_path,:notice=>"Thanksforyourmessage,Iwillbeintouchsoon")elseflash[:error]

  5. ruby-on-rails - 在 RSpec 中,如何以任意顺序期望具有不同参数的多条消息? - 2

    RSpec似乎按顺序匹配方法接收的消息。我不确定如何使以下代码工作:allow(a).toreceive(:f)expect(a).toreceive(:f).with(2)a.f(1)a.f(2)a.f(3)我问的原因是a.f的一些调用是由我的代码的上层控制的,所以我不能对这些方法调用添加期望。 最佳答案 RSpecspy是测试这种情况的一种方式。要监视一个方法,用allowstub,除了方法名称之外没有任何约束,调用该方法,然后expect确切的方法调用。例如:allow(a).toreceive(:f)a.f(2)a.f(1)

  6. ruby-on-rails - 设计注册确认 - 2

    我在我的项目中有一个用户和一个管理员角色。我使用Devise创建了身份验证。在我的管理员角色中,我没有任何确认。在我的用户模型中,我有以下内容:devise:database_authenticatable,:confirmable,:recoverable,:rememberable,:trackable,:validatable,:timeoutable,:registerable#Setupaccessible(orprotected)attributesforyourmodelattr_accessible:email,:username,:prename,:surname,:

  7. ruby-on-rails - 闪存消息存储在哪里? - 2

    我以为它们存储在cookie中-但不,检查cookie没有任何结果。session也不存储它们。那么,我在哪里可以找到它们?我需要这个来直接设置它们(而不是通过flashhash)。 最佳答案 它们存储在inyoursessionstore.自rails2.0以来的默认设置是cookie存储,但请检查config/initializers/session_store.rb以检查您是否使用默认设置以外的东西。 关于ruby-on-rails-闪存消息存储在哪里?,我们在StackOverf

  8. Ruby SSL 错误 - sslv3 警报意外消息 - 2

    我正在尝试在ruby​​脚本中连接到服务器https://www.xpiron.com/schedule。但是,当我尝试连接时:require'open-uri'doc=open('https://www.xpiron.com/schedule')我收到以下错误消息:OpenSSL::SSL::SSLError:SSL_connectreturned=1errno=0state=SSLv2/v3readserverhelloA:sslv3alertunexpectedmessagefrom/usr/local/lib/ruby/1.9.1/net/http.rb:678:in`conn

  9. ruby-on-rails - Ruby on Rails - 需要在每周的特定时间将消息发送到电子邮件 - 2

    我想知道我应该如何着手这个项目。我需要每周向人们发送一次电子邮件。但是,这必须在每周的特定时间自动生成并发送。编码有多难?我需要知道是否有任何书籍可以提供帮助,或者你们中的任何人是否可以指导我。它必须使用ruby​​onrails进行编程。因此有一个网络服务和数据库集成。干杯 最佳答案 为什么这么复杂?您只需安排工作。您可以使用Delayed::Job例如。Delayed::Job让您可以使用run_at符号在特定时间安排作业,如下所示:Delayed::Job.enqueue(SendEmailJob.new(...),:run_

  10. ruby-on-rails - 使用 ruby​​ on rails 在 json 中发送错误消息 - 2

    我正在验证ruby​​onrails中的输入字段。我检查用户是否输入或填写了这些字段。如果假设name字段未填写,则向用户发送一条错误消息,指示name字段未填写。其他错误也是如此。我如何使用ruby​​onrails在json中发送这种消息。这是我现在正在做的。这个模型validates:email,:name,:company,:presence=>truevalidates_format_of:email,:with=>/\A[a-z0-9!#\$%&'*+\/=?^_`{|}~-]+(?:\.[a-z0-9!#\$%&'*+\/=?^_`{|}~-]+)*@(?:[a-z0-9

随机推荐