【Rabbitmq篇】RabbitMQ⾼级特性----消息确认

• ⾃动确认

当autoAck等于true时, RabbitMQ 会⾃动把发送出去的消息置为确认,然后从内存(或者磁盘)中删除,⽽不管消费者是否真正地消费到了这些消息.⾃动确认模式适合对于消息可靠性要求不⾼的场景.

• ⼿动确认

当autoAck等于false时，RabbitMQ会等待消费者显式地调⽤Basic.Ack命令,回复确认信号后才从内存(或者磁盘)中移去消息.这种模式适合对消息可靠性要求⽐较⾼的场景.

手动确认方法又分为三种：

肯定确认:Channel.basicAck(long deliveryTag, boolean multiple) RabbitMQ已知道该消息并且成功的处理消息，可以将其丢弃了.
否定确认: Channel.basicReject(long deliveryTag, boolean requeue)
消费者客⼾端可以调⽤channel.basicReject⽅法来告诉RabbitMQ拒绝这个消息.
否定批量确认: Channel.basicNack(long deliveryTag, boolean multiple,boolean requeue)
Basic.Reject命令⼀次只能拒绝⼀条消息,如果想要批量拒绝消息,则可以使⽤Basic.Nack这个命令.消费者客⼾端可以调⽤channel.basicNack⽅法来实现.

参数说明：

1）deliveryTag ：
消息的唯⼀标识，它是⼀个单调递增的64位的⻓整型值. deliveryTag 是每个通道
（Channel）独⽴维护的,所以在每个通道上都是唯⼀的.当消费者确认(ack)⼀条消息时,必须使⽤对应的通道上进⾏确认.

2）multiple

是否批量确认.在某些情况下,为了减少⽹络流量,可以对⼀系列连续的 deliveryTag 进
⾏批量确认.值为true则会⼀次性ack所有⼩于或等于指定deliveryTag的消息.值为false,则只确认当前指定deliveryTag的消息.

3）requeue

表⽰拒绝后,这条消息如何处理.如果requeue参数设置为true,则RabbitMQ会重新将这条
消息存⼊队列,以便可以发送给下⼀个订阅的消费者.如果requeue参数设置为false,则RabbitMQ会把消息从队列中移除,⽽不会把它发送给新的消费者.

二. 代码实现（spring环境）

1.可以直接使用RabbitMQ Java Client 库

2.使用spring集成的amqp

主要介绍第二种，在spring环境下实现

Spring-AMQP 对消息确认机制提供了三种策略.

public enum AcknowledgeMode { 
    NONE //确认,
    MANUAL//手动 ,
    AUTO //默认;
}

配置相关信息：

基本信息以及确认机制

队列，交换机，以及它们之间的绑定关系

package com.bite.extensions.config;

import com.bite.extensions.constant.Constants;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
    @Bean("ackQueue")
    public Queue ackQueue() {
        return QueueBuilder.durable(Constants.ACK_QUEUE).build();
    }
    @Bean("directExchange")
    public DirectExchange directExchange() {
        return ExchangeBuilder.directExchange(Constants.ACK_EXCHANGE).build();
    }
    @Bean("ackBinding")
    public Binding ackBinding(@Qualifier("directExchange") DirectExchange directExchange,@Qualifier("ackQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(directExchange).with("ack");
    }
}

生产者：

主要解释消费者在不同确认机制的状态

package com.bite.extensions.controller;

import com.bite.extensions.constant.Constants;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RequestMapping("/producer")
@RestController
public class ProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/ack")
    public String ack() {
        rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,"ack","consumer ack mode test...");
        return "消息发送成功!";
    }
}

1）AcknowledgeMode.NONE

这种模式下,消息⼀旦投递给消费者,不管消费者是否成功处理了消息,RabbitMQ就会⾃动确认
消息,从RabbitMQ队列中移除消息.如果消费者处理消息失败,消息可能会丢失.

1）消费者正常消费情况下

package com.bite.extensions.listener;

import com.bite.extensions.constant.Constants;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void handleMessage(Message message, Channel channel) throws UnsupportedEncodingException {
        //消费者逻辑
        long deliverTag = message.getMessageProperties().getDeliveryTag();
        System.out.printf("接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
        //业务逻辑处理
        System.out.println("业务逻辑处理！");

        System.out.println("业务逻辑完成！");
    }
}

消费者正确处理,MQ删除相应信息

2）消费者异常消费情况下

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void handleMessage(Message message, Channel channel) throws UnsupportedEncodingException {
        //消费者逻辑
        long deliverTag = message.getMessageProperties().getDeliveryTag();
        System.out.printf("接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
        //业务逻辑处理
        System.out.println("业务逻辑处理！");
        int num = 3/0; //异常
        System.out.println("业务逻辑完成！");
    }
}

可以看到,消费者处理失败,但是消息已经从RabbitMQ中移除.

2 ）AcknowledgeMode.AUTO

这种模式下,消费者在消息处理成功时会⾃动确认消息,但如果处理过程中抛出了异常,则不会确认消息.

    listener:
      simple:
        acknowledge-mode: auto  #消息接收确认

1）消费者正常消费情况下

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void handleMessage(Message message, Channel channel) throws UnsupportedEncodingException {
        //消费者逻辑
        long deliverTag = message.getMessageProperties().getDeliveryTag();
        System.out.printf("接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
        //业务逻辑处理
        System.out.println("业务逻辑处理！");
        //int num = 3/0;
        System.out.println("业务逻辑完成！");
    }
}

2）消费者异常消费情况下

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void handleMessage(Message message, Channel channel) throws UnsupportedEncodingException {
        //消费者逻辑
        long deliverTag = message.getMessageProperties().getDeliveryTag();
        System.out.printf("接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
        //业务逻辑处理
        System.out.println("业务逻辑处理！");
        int num = 3/0;
        System.out.println("业务逻辑完成！");
    }
}

..........
接收到信息: consumer ack mode test..., deliveryTag: 88
业务逻辑处理！
2024-11-17T15:19:11.420+08:00  WARN 22936 --- [rabbitmq-extensions-demo] [ntContainer#0-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
接收到信息: consumer ack mode test..., deliveryTag: 89
业务逻辑处理！
2024-11-17T15:19:11.477+08:00  INFO 22936 --- [rabbitmq-extensions-demo] [ntContainer#0-2] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
2024-11-17T15:19:11.477+08:00  INFO 22936 --- [rabbitmq-extensions-demo] [ntContainer#0-2] o.s.a.r.l.SimpleMessageListenerContainer : Successfully waited for workers to finish.

消费者处理异常，会一直重试发送，所有仍然保留在mq中

3）AcknowledgeMode.MANUAL

⼿动确认模式下,消费者必须在成功处理消息后显式调⽤ basicAck ⽅法来确认消息.如果消
息未被确认,RabbitMQ会认为消息尚未被成功处理,并且会在消费者可⽤时重新投递该消息,这
种模式提⾼了消息处理的可靠性,因为即使消费者处理消息后失败,消息也不会丢失,⽽是可以被重新处理.

    listener:
      simple:
        acknowledge-mode: manual#消息接收确认

1）消费者正常消费情况下

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void handleMessage(Message message, Channel channel) throws Exception {
        //消费者逻辑
        long deliverTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.printf("接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
            //业务逻辑处理
            System.out.println("业务逻辑处理！");
            //int  num = 3/0;
            System.out.println("业务逻辑完成！");
            //肯定确认
            channel.basicAck(deliverTag,false);
        } catch (Exception e) {
            //否定确认
           channel.basicNack(deliverTag,false,true);
        }
    }
}

如果不进行确认又会发送什么？

当我们使用手动确认（manual）的时候，一定要手动添加上肯定确认，不然即使消费者处理成功，也不会进行确认！

2）消费者异常消费情况下

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void handleMessage(Message message, Channel channel) throws Exception {
        //消费者逻辑
        long deliverTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.printf("接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
            //业务逻辑处理
            System.out.println("业务逻辑处理！");
            int  num = 3/0;
            System.out.println("业务逻辑完成！");
            //肯定确认
            channel.basicAck(deliverTag,false);
        } catch (Exception e) {
            //否定确认
           channel.basicNack(deliverTag,false,true);
        }
    }
}

否定确认完，又会进行重新入队，会变成Ready状态

此时修改为false，不让它入队，会发生什么？

消费者处理异常，会不停的重试

使用manual，一定要进行手动确认

总结：

模式	确认方式	可靠性	性能	使用场景
None	无确认	低，可能丢失消息	高	不关心消息是否成功消费，丢失消息可容忍的场景
Auto	自动确认	较低，可能丢失消息	较高	对丢失消息容忍度较高的场景
Manual	手动确认	高，消息只有成功处理才会确认	较低	需要确保每条消息被成功消费的场景

None 适用于性能要求高，但对消息丢失不敏感的场景。
Auto 适合那些不需要太高消息可靠性的应用，但仍然需要自动化确认机制。
Manual 最适合那些对消息处理的可靠性要求较高，尤其是在出现异常时需要精细控制消息是否重新入队或丢弃的场景。

选择哪种模式取决于你的具体需求，尤其是对于消息可靠性的要求以及系统的性能考虑。

结语： 写博客不仅仅是为了分享学习经历，同时这也有利于我巩固知识点，总结该知识点，由于作者水平有限，对文章有任何问题的还请指出，接受大家的批评，让我改进。同时也希望读者们不吝啬你们的点赞+收藏+关注，你们的鼓励是我创作的最大动力！

原文地址：https://blog.csdn.net/chaodddddd/article/details/143831131

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：计算机组成原理笔记----基础篇
下一篇：leetcode刷题记录（三十一）——225. 用队列实现栈

网络协议之Ethernet
Ethernet协议作为局域网技术的核心，自1970年代诞生以来，经历了从10Mbps到更高速率的不断演进。它采用CSMA/CD协议管理数据传输，支持多种拓扑结构，以星型为主。以太网凭借其高速率、高稳
阅读更多2024-11-18
解析流式处理技术在实时数据ETL的具体应用
流处理技术可以对这些数据流进行初步筛选，只将有价值的数据传递给后续的分析系统，从而减轻后续系统的负担，提高整体的处理效率。例如，在网络日志数据处理中，流处理系统可以先过滤掉一些无关的日志信息（如一些常
阅读更多2024-11-18
【JavaEE进阶】CSS
CSS 能够对⽹⻚中元素位置的排版进⾏像素级精确控制, 实现美化⻚⾯的效果. 能够做到⻚⾯的样式和结构分离.⼀个类可以被多个标签使⽤, ⼀个标签也能使⽤多个类(多个类名要使⽤空格分割, 这种做法可以让
阅读更多2024-11-18
【Linux】Socket编程基础
socket编程预备知识
阅读更多2024-11-18
C# 中的异常处理：编写健壮的应用程序
异常是在程序执行过程中发生的一种非正常情况。当程序无法正常执行某段代码时，会抛出一个异常对象。异常对象包含了有关错误的详细信息，如错误类型、错误消息和堆栈跟踪等。
阅读更多2024-11-18
npm : 无法加载文件 C:\Program Files\nodejs\npm.ps1，因为在此系统上禁止运行脚本。
RemoteSigned 就可以实现访问。
阅读更多2024-11-18
【代码pycharm】动手学深度学习v2-04 数据操作 + 数据预处理
老师课上用的jupyter我用pycharm实现的
阅读更多2024-11-18
【HarmonyOS】鸿蒙系统在租房项目中的项目实战（二）
从今天开始，博主将开设一门新的专栏用来讲解市面上比较热门的技术 “鸿蒙开发”，对于刚接触这项技术的小伙伴在学习鸿蒙开发之前，有必要先了解一下鸿蒙，从你的角度来讲，你认为什么是鸿蒙呢？它出现的意义又是什
阅读更多2024-11-18
删除缓存之后，浏览器显示登录新设备
当C盘缓存被清除时，这些存储的Token也会被删除，导致浏览器无法识别之前的登录状态，因此需要重新登录。清除缓存后，如果Refresh Token也丢失，用户可能需要重新登录以获取新的Token。综上
阅读更多2024-11-18
Java Function 的妙用：化繁为简的魔法师！
在 Java 开发中，Function是一个非常重要的函数式接口，它的存在让代码从繁琐走向简洁，从普通迈向优雅。今天，我们就来聊聊Function的那些“妙用”，以及它如何帮助开发者写出更简洁、更强大
阅读更多2024-11-18

【Rabbitmq篇】RabbitMQ⾼级特性----消息确认

前言：

一.消息确认机制

• ⾃动确认

• ⼿动确认

手动确认方法又分为三种：

二. 代码实现（spring环境）

配置相关信息：

1）AcknowledgeMode.NONE

2 ）AcknowledgeMode.AUTO

3）AcknowledgeMode.MANUAL

总结：

相关文章