zoukankan      html  css  js  c++  java
  • RabbitMQ的消息确认机制

    一:确认种类

    RabbitMQ的消息确认有两种。

    一种是消息发送确认。这种是用来确认生产者将消息发送给交换器,交换器传递给队列的过程中,消息是否成功投递。发送确认分为两步,一是确认是否到达交换器,二是确认是否到达队列。

    第二种是消费接收确认。这种是确认消费者是否成功消费了队列中的消息。

    二:消息发送确认

    (1)ConfirmCallback

    通过实现ConfirmCallBack接口,消息发送到交换器Exchange后触发回调。

    RabbitMQ的消息确认机制

     

    使用该功能需要开启确认,spring-boot中配置如下:

    spring.rabbitmq.publisher-confirms = true

    (2)ReturnCallback

    通过实现ReturnCallback接口,如果消息从交换器发送到对应队列失败时触发(比如根据发送消息时指定的routingKey找不到队列时会触发)

    RabbitMQ的消息确认机制

     

    使用该功能需要开启确认,spring-boot中配置如下:

    spring.rabbitmq.publisher-returns = true

    三:消息接收确认

    (1)确认模式

    • AcknowledgeMode.NONE:不确认
    • AcknowledgeMode.AUTO:自动确认
    • AcknowledgeMode.MANUAL:手动确认

    spring-boot中配置方法:

    spring.rabbitmq.listener.simple.acknowledge-mode = manual

    (2)手动确认

    RabbitMQ的消息确认机制

    未确认的消息数

    上图为channel中未被消费者确认的消息数。

    通过RabbitMQ的host地址加上默认端口号15672访问管理界面。

    (2.1)成功确认

    void basicAck(long deliveryTag, boolean multiple) throws IOException;

    deliveryTag:该消息的index

    multiple:是否批量. true:将一次性ack所有小于deliveryTag的消息。

    消费者成功处理后,调用channel.basicAck(message.getMessageProperties().getDeliveryTag(), false)方法对消息进行确认。

    (2.2)失败确认

    void basicNack(long deliveryTag, boolean multiple, boolean requeue)

    throws IOException;

    deliveryTag:该消息的index。

    multiple:是否批量. true:将一次性拒绝所有小于deliveryTag的消息。

    requeue:被拒绝的是否重新入队列。

    void basicReject(long deliveryTag, boolean requeue) throws IOException;

    deliveryTag:该消息的index。

    requeue:被拒绝的是否重新入队列。

    channel.basicNack 与 channel.basicReject 的区别在于basicNack可以批量拒绝多条消息,而basicReject一次只能拒绝一条消息。

    消息队列的基础知识可以参考:消息队列RabbitMQ基础知识详解

    四:思考

    (1)手动确认模式,消息手动拒绝中如果requeue为true会重新放入队列,但是如果消费者在处理过程中一直抛出异常,会导致入队-》拒绝-》入队的循环,该怎么处理呢?

    第一种方法是根据异常类型来选择是否重新放入队列。

    第二种方法是先成功确认,然后通过channel.basicPublish()重新发布这个消息。重新发布的消息网上说会放到队列后面,进而不会影响已经进入队列的消息处理。

    void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)

    throws IOException;

    (2)消息确认的作用是什么?

    为了防止消息丢失。消息丢失分为发送丢失和消费者处理丢失,相应的也有两种确认机制。

    本文转自:https://www.toutiao.com/a6583957771840913934/?tt_from=mobile_qq&utm_campaign=client_share&timestamp=1532999387&app=news_article&utm_source=mobile_qq&iid=39062783162&utm_medium=toutiao_android

  • 相关阅读:
    leetcode 190 Reverse Bits
    vs2010 单文档MFC 通过加载位图文件作为客户区背景
    leetcode 198 House Robber
    记忆化搜索(DP+DFS) URAL 1183 Brackets Sequence
    逆序数2 HDOJ 1394 Minimum Inversion Number
    矩阵连乘积 ZOJ 1276 Optimal Array Multiplication Sequence
    递推DP URAL 1586 Threeprime Numbers
    递推DP URAL 1167 Bicolored Horses
    递推DP URAL 1017 Staircases
    01背包 URAL 1073 Square Country
  • 原文地址:https://www.cnblogs.com/nizuimeiabc1/p/9397326.html
Copyright © 2011-2022 走看看