RabbitMQ 发布确认机制

这篇具有很好参考价值的文章主要介绍了RabbitMQ 发布确认机制。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

发布确认模式是避免消息由生产者到RabbitMQ消息丢失的一种手段

原理说明

  生产者通过调用channel.confirmSelect方法将信道设置为confirm模式,之后RabbitMQ会返回Confirm.Select-OK命令表示同意生产者将当前信道设置为confirm模式。
  confirm模式下的信道所发送的消息都将被应带ack或者nack一次,不会出现一条消息即被ack又被nack的情况,并且RabbitMQ也并没有对消息被confirm的快慢做出保证,消息被confirm是异步进行。
RabbitMQ 发布确认机制,中间件 - RabbitMQ,rabbitmq,分布式

  如上图所示为confirm模式下的消息发送过程,其中4和6为异步应答,也就是说4过程并不一定在5之前,也有可能是在下一条消息发送后才会进行上一条消息的应答。
  RabbitMQ 事务和发送确认机制确保的是消息能够正确的发送至RabbitMQ的交换机,如果交换机没有匹配的队列,那么消息也会被丢失。和事务不同的是,发布确认机制是异步进行的,因此在性能上发布确认模式将更加优秀,需要注意的是:事务和确认机制是互斥的,不能共存
  事务机制和发布确认机制都存在以下注意点:

  • 如果消息需要持久化并且存在队列,则在消息入队并且持久化后进行返回事务提交成功或者应答消息。
  • 如果消息不需要持久化但是存在队列,则在消息入队后返回事务提交成功或者应答消息。
  • 如果消息不可路由到队列中,则在路由失败后返回事务提交成功或者应答消息。

  上文中一直强调的时发布确认针对发布发送到RabbitMQ中的交换机进行保证,但消息实际是否能入队发布确认机制并不能提供保证,因此还需要和mandatory参数配合使用。

实现方式

  RabbitMQ的发布确认机制可以分为三种实现方式:阻塞等待确认、批量阻塞等待确认、异步确认。
阻塞等待确认:每当消息发送后,发送者都阻塞的等待应答消息。这种实现方式将无法体现发布确认模式的异步性能优势。
批量阻塞确认:批量阻塞确认类似于阻塞等待确认,区别在于批量阻塞确认并不会针对每条消息进行阻塞等待,他会针对一些消息进行统一阻塞等待应答消息。这种实现方式将同步和异步结合起来进行使用,对应答性能有一定的提升。
异步应答:实现一个监听器的方式接收应答消息,应答消息的处理逻辑不会影响消息的发送,消息的应答和消息发送是异步进行的,他们并不直接相互干扰。
上面对三种确认方式进行简单说明,下面将分别介绍发布确认机制的实现方式。

开启confirm(确认)模式

  确认模式的开启是针对信道设置的,一旦信道进入了confirm模式,所有在该信道上面发布的消息都会被指派唯一的ID,RabbitMQ也将针对该信道发送的所有消息都进行应答。
  RabbitMQ回传给生产者的确认消息中的deliverryTag包含了确认消息的序号,但在使用(批量)阻塞确认方式进行实现的时候该消息序号无意义。开启confirm模式仅需要以下代码进行实现即可:

channel.confirmSelect();

阻塞确认

  阻塞确认的方式依赖于channel.waitForConfirms()方法,该方法如下所示:

    /**
     * Wait until all messages published since the last call have been
     * either ack'd or nack'd by the broker.  Note, when called on a
     * non-Confirm channel, waitForConfirms throws an IllegalStateException.
     * @return whether all the messages were ack'd (and none were nack'd)
     * @throws java.lang.IllegalStateException
     */
    boolean waitForConfirms() throws InterruptedException;

  自从上次调用该方法后直到所有发送的消息都被应答后返回所有消息的应答结果,如果所有发送的消息应答结果都是成功则返回true,一旦存在任何一条消息应答失败则返回false。
  根据该方法的描述可知,可以通过该方法实现阻塞等待确认和批量阻塞确认两种方案,区别仅在于是发送一条消息调用一次该方法还是发送一批消息后调用一次这个方法。
  阻塞等待确认的方式如下代码所示:

//发送消息
channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
// 如果发送失败则进行该条消息的重新发送
if(!channel.waitForConfirms()){
    channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
}

 阻塞批量确认的方式如下代码所示:

        // 存储未应带消息队列
        List<String> messages = new ArrayList<>();
        for (int i = 1; i < 20000 ;  i++){
            String msg = String.valueOf(i);
            messages.add(msg);
            channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());
            // 每发送十条消息进行一次确认
            if(i > 0 && i % 10 == 0 ){
                // 如果确认不通过则将消息重新发送
                if(!channel.waitForConfirms()){
                    for (String e : messages) {
                        channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,e.getBytes());
                    }
                }else{
                    // 如果确认成功则将这些消息从未应答队列中移除
                    messages.clear();
                }
            }
        }

异步确认

  客户端Channel提供了addConfirmListener方法,该可以添加ConfirmListener这个回调接口,该接口包含两个方法:handleAck和handleNack,分别用来处理饭hi的Ack和Nack,这两个方法都将返回一个参数deliveryTag(消息的唯一有序序号)和一个boolean型参数multiple,如果该参数为true表示自该消息之前的所有消息RabbitMQ服务都已经做出了应答。我们可以通过该值实现具体业务的发布确认。

/**
* Implement this interface in order to be notified of Confirm events.
* Acks represent messages handled successfully; Nacks represent
* messages lost by the broker.  Note, the lost messages could still
* have been delivered to consumers, but the broker cannot guarantee
* this.
* For a lambda-oriented syntax, use {@link ConfirmCallback}.
*/
public interface ConfirmListener {
  void handleAck(long deliveryTag, boolean multiple)
      throws IOException;

  void handleNack(long deliveryTag, boolean multiple)
      throws IOException;
}

  异步确认的方式实现起来比较复杂,在生产者端需要维护一个消息队列,如果消息应答成功则将该消息从队列中移除,如果消息应答失败则将该消息再重新发送或进行其他业务处理。该逻辑伪码如下所示:

        // 存储未确认消息,其中key为消息序号,value为消息实体
        HashMap<Long,String> msgMap = new HashMap<>();
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                msgMap.remove(deliveryTag);
            }

            /**
             * 如果消息应带结果为nack则重新发送该消息
             * @param deliveryTag
             * @param multiple
             * @throws IOException
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                String msg = msgMap.get(deliveryTag);
                if(msg != null){
                    channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());
                }
            }
        });
        for (int i = 1; i < 20000 ;  i++){
            String msg = String.valueOf(i);
            // 将消息序号和消息存储map中
            msgMap.put(channel.getNextPublishSeqNo(),msg);
            channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());
        }

  上述代码使用了map存储消息序号和消息实体,这种存储方式应该会存在风险,由于监听器和消息发送过程是异步进行了,因此可能会存在线程安全的问题,HashMap是非线程安全的。

总结

  发布确认模式是为我们解决消息自生产者发送到RabbitMQ交换机过程中消息丢失的问题的,这一场景需求我们也可以通过事务机制实现。发布确认模式和事务机制比较如下表所示:

比较 事务机制 发布确认机制
实现方式 通过AMQP协议层面实现 轻量级实现,采用RabbitMQ应答机制
命令详解 Tx.Select
Basic.Publish
Tx.Commit
Commit.OK
Basic.Publish
Basic.Ack
性能 同步,性能较慢 可异步实现也可同步实现,性能快,AMQP命令交互少
消息到达队列时机 事务提交后消息才会进入队列,消息入队存在滞后性 消息发送后就进入队列,发布确认模式不影响消息进入队列时机
事务提交成功或消息应答时机 消息被交换机处理完成后,或消息不可达 同事务
实现复杂度 简单 相对复杂
适合场景 批量发送消息,实现批量消息的原子性和一致性 确保消息发送到交换机

  发布确认模式的具体实现可以划分为三种:阻塞等待、批量确认、异步确认,这三者的比较如下表所示:

比较内容 阻塞等待 批量等待 异步确认
性能
实现复杂度
确认范围 每条消息 批量消息 每条消息
是否可以精准确认每条消息

  根据上述内容,我们在实现避免消息自生产者到交换机丢失的机制时建议使用发布确认模式的异步确认,因为异步确认性能最高,并且可以准确的得到被应答的消息的序号,有助于我们进行后续逻辑处理。文章来源地址https://www.toymoban.com/news/detail-638417.html

到了这里,关于RabbitMQ 发布确认机制的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

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

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

相关文章

  • 消息中间件之RabbitMQ

    1.基于AMQP协议Erlang语言开发的一款消息中间件,客户端语言支持比较多, 比如Python,Java,Ruby,PHP,JS,Swift.运维简单,灵活路由,但是性能不高, 可以满足一般场景下的业务需要,三高场景下吞吐量不高,消息持久化没有采取 零拷贝技术,消息堆积时,性能会下降 2.消息吞吐量在

    2024年01月19日
    浏览(86)
  • 中间件RabbitMQ消息队列介绍

    1.1 什么是 MQ MQ ( message queue ),从字面意思上看,本质是个队列, FIFO 先入先出,只不过队列中存放的内容是 message 而已,还是一种跨进程的通信机制,用于上下游传递消息。在互联网架构中, MQ 是一种非常常 见的上下游 逻辑解耦+物理解耦 的消息通信服务。使用了 MQ 之

    2024年02月13日
    浏览(72)
  • RabbitMQ 消息中间件 消息队列

    RabbitMQ 1、RabbitMQ简介 RabbiMQ是⽤Erang开发的,集群⾮常⽅便,因为Erlang天⽣就是⼀⻔分布式语⾔,但其本身并 不⽀持负载均衡。支持高并发,支持可扩展。支持AJAX,持久化,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。 2、RabbitMQ 特点 可

    2024年02月03日
    浏览(69)
  • 【RabbitMQ教程】前言 —— 中间件介绍

                                                                       💧 【 R a b b i t M Q 教程】前言——中间件介绍 color{#FF1493}{【RabbitMQ教程】前言 —— 中间件介绍} 【 R abbi tMQ 教程】前言 —— 中间件介绍 💧           🌷 仰望天空,妳

    2024年02月08日
    浏览(70)
  • 高性能消息中间件 RabbitMQ

    消息队列 MQ全称Message Queue(消息队列),是在消息的传输过程中保存消息的容器。多用于系统之间的 异步通信 。 同步通信相当于两个人当面对话,你一言我一语。必须及时回复: 异步通信相当于通过第三方转述对话,可能有消息的延迟,但不需要二人时刻保持联系。 消息

    2024年02月11日
    浏览(108)
  • Springboot整合RabbitMQ消息中间件

    spring-boot-rabbitmq–消息中间件整合 前言:RabbitMQ的各种交换机说明 1、直连交换机 生产者发布消息时必须带着routing-key,队列绑定到交换机时必须指定binding-key ,且routing-key和binding-key必须完全相同,如此才能将消息路由到队列中 直连交换机通常用来循环分发任务给多个workers,

    2024年02月11日
    浏览(47)
  • 消息队列中间件(二)- RabbitMQ(一)

    接收,存储,转发消息 生产者 交换机 队列 消费者 简单模式 工作模式 发布 路由模式 主题模式 发布订阅模式 Broker 接收和分发消息的应用 Virtual host 虚拟分组 Connection: TCP连接 Channel: 节省连接,每次访问建立一次Connection消耗太大,所以使用信道代替连接 交换机 队列 www.r

    2024年02月11日
    浏览(68)
  • Python学习(3)- 消息中间件rabbitmq

    消息队列中间件是分布式系统中重要的组件,主要解决 应用解耦 , 异步消息 , 流量削锋 等问题,实现高性能,高可用,可伸缩和最终一致性架构 rabbitmq 简单模式 生产者: 链接rabbitmq 创建队列 向指定的队列中插入数据 交换机模式 发布订阅模式 模式 模糊匹配模式

    2024年02月22日
    浏览(51)
  • 「中间件」rabbitmq 消息队列基础知识

    RabbitMQ是一个消息队列软件,用于在应用程序之间转发消息。以下是RabbitMQ的基本概念: 消息:RabbitMQ中的消息是传递的基本单位,它由消息头和消息体组成。 队列(Queue):队列是消息的缓冲区,用于存储待处理的消息。 交换器(Exchange):交换器是接收生产者发送的消息并

    2024年02月07日
    浏览(62)
  • 中间件_RabbitMQ五种消息模型

    RabbitMQ官方文档 RabbitMQ 提供了5种常用消息模型。但是其实3、4、5这三种都属于订阅模型,只不过进行路由的方式不同。 简单消息队列官方文档 1、创建简单消息队列 2、导入依赖 3、编写生产者测试类SpringAmqpTest,并利用 RabbitTemplate 实现消息发送 4、编写消费者,监听队列消息

    2024年02月06日
    浏览(43)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包