详解SpringBoot整合RabbitMQ如何实现消息确认
简介
本文介绍SpringBoot整合RabbitMQ如何进行消息的确认。
生产者消息确认
介绍
发送消息确认：用来确认消息从 producer发送到 broker 然后broker 的 exchange 到 queue过程中，消息是否成功投递。
如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出；如果是镜像队列，所有镜像接受成功后发确认消息。
流程
- 如果消息没有到达exchange，则confirm回调，ack=false
- 如果消息到达exchange，则confirm回调，ack=true
- exchange到queue成功，则不回调return
- exchange到queue失败，则回调return(需设置mandatory=true，否则不会回调，这样消息就丢了)
配置
application.yml
# 发送者开启 confirm 确认机制 spring.rabbitmq.publisher-confirms=true # 发送者开启 return 确认机制 spring.rabbitmq.publisher-returns=true
ConfirmCallback
ConfirmCallback：消息只要被 RabbitMQ broker 接收到就会触发confirm方法。
@Slf4j
@Component
public class ConfirmCallbackService implements RabbitTemplate.ConfirmCallback {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (!ack) {
log.error("confirm==>发送到broker失败\r\n" +
"correlationData={}\r\n" + "ack={}\r\n" + "cause={}",
correlationData, ack, cause);
} else {
log.info("confirm==>发送到broker成功\r\n" +
"correlationData={}\r\n" + "ack={}\r\n" + "cause={}",
correlationData, ack, cause);
}
}
}
correlationData：对象内部有id （消息的唯一性）和Message。
（若ack为false，则Message不为null，可将Message数据 重新投递；若ack是true，则correlationData为null）
ack：消息投递到exchange 的状态，true表示成功。
cause：表示投递失败的原因。 （若ack为false，则cause不为null；若ack是true，则cause为null）
给每一条信息添加一个dataId，放在CorrelationData，这样在RabbitConfirmCallback返回失败时可以知道哪个消息失败。
public void send(String dataId, String exchangeName, String rountingKey, String message){
CorrelationData correlationData = new CorrelationData();
correlationData.setId(dataId);
rabbitTemplate.convertAndSend(exchangeName, rountingKey, message, correlationData);
}
public String receive(String queueName){
return String.valueOf(rabbitTemplate.receiveAndConvert(queueName));
}
2.1版本开始，CorrelationData对象具有ListenableFuture，可用于获取结果，而不是在rabbitTemplate上使用ConfirmCallback。
CorrelationData cd1 = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("exchange", queue.getName(), "foo", cd1);
assertTrue(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck());
ReturnCallback
ReturnCallback：如果消息未能投递到目标 queue 里将触发returnedMessage方法。
若向 queue 投递消息未成功，可记录下当前消息的详细投递数据，方便后续做重发或者补偿等操作。
注意：需要rabbitTemplate.setMandatory(true);
当mandatory设置为true时，若exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法将消息返还给生产者。当mandatory设置为false时，出现上述情况broker会直接将消息丢弃。
代码：
@Slf4j
@Component
public class ReturnCallbackService implements RabbitTemplate.ReturnCallback {
@Override
public void returnedMessage(Message message, int replyCode, String replyText,
String exchange, String routingKey) {
log.info("returnedMessage==> \r\n" + "message={}\r\n" + "replyCode={}\r\n" +
"replyText={}\r\n" + "exchange={}\r\n" + "routingKey={}",
message, replyCode, replyText, exchange, routingKey);
}
}
message（消息体）、replyCode（响应code）、replyText（响应内容）、exchange（交换机）、routingKey（队列）。
注册ConfirmCallback和ReturnCallback
整合后的写法
package com.example.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
@Slf4j
@Configuration
public class RabbitCallbackConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
@Bean
RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
return rabbitTemplate;
}
// 下边这样写也可以
// @Autowired
// private RabbitTemplate rabbitTemplate;
// @PostConstruct
// public void init() {
// rabbitTemplate.setMandatory(true);
// rabbitTemplate.setReturnCallback(this);
// rabbitTemplate.setConfirmCallback(this);
// }
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (!ack) {
log.error("confirm==>发送到broker失败\r\n" +
"correlationData={}\r\n" + "ack={}\r\n" + "cause={}",
correlationData, ack, cause);
} else {
log.info("confirm==>发送到broker成功\r\n" +
"correlationData={}\r\n" + "ack={}\r\n" + "cause={}",
correlationData, ack, cause);
}
}
@Override
public void returnedMessage(Message message, int replyCode, String replyText,
String exchange, String routingKey) {
log.info("returnedMessage==> \r\n" + "message={}\r\n" + "replyCode={}\r\n" +
"replyText={}\r\n" + "exchange={}\r\n" + "routingKey={}",
message, replyCode, replyText, exchange, routingKey);
}
}
消费者消息确认
介绍
| 确认方式 | 简介 | 详述 |
|---|---|---|
| auto（默认） | 根据消息消费的情况，智能判定 | 若消费者抛出异常，则mq不会收到确认消息，mq会一直此消息发出去 若消费者没有抛出异常，则mq会收到确认消息，mq不会再次将此消息发出去。 若消费者在消费时所在服务挂了，mq不会再次将此消息发出去。 |
| none | mq发出消息后直接确认消息 | |
| manual | 消费端手动确认消息 | 消费者调用 ack、nack、reject 几种方法进行确认，可以在业务失败后进行一些操作，如果消息未被 ACK 则消息还会存在于MQ，mq会一直将此消息发出去。 如果某个服务忘记 ACK 了，则 RabbitMQ 不会再发送数据给它，因为 RabbitMQ 认为该服务的处理能力有限。 |
只要消息没有被消费者确认（包括没有自动确认），会导致消息一直被失败消费，死循环导致消耗大量资源。正确的处理方式是：发生异常，将消息记录到db，再通过补偿机制来补偿消息，或者记录消息的重复次数，进行重试，超过几次后再放到db中。
消息确认三种方式配置方法
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.direct.acknowledge-mode=manual
手动确认三种方式
basicAck，basicNack，basicReject
basicAck
含义
表示成功确认，使用此回执方法后，消息会被RabbitMQ broker 删除。
函数原型
void basicAck(long deliveryTag, boolean multiple)
deliveryTag
- 消息投递序号
- 每次消费消息或者消息重新投递后，deliveryTag都会增加。手动消息确认模式下，我们可以对指定deliveryTag的消息进行ack、nack、reject等操作。
multiple
- 是否批量确认
- 值为 true 则会一次性 ack所有小于当前消息 deliveryTag 的消息。
示例： 假设我先发送三条消息deliveryTag分别是5、6、7，可它们都没有被确认，当我发第四条消息此时deliveryTag为8，multiple设置为 true，会将5、6、7、8的消息全部进行确认。
实例
@RabbitHandler
public void process(String content, Channel channel, Message message){
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
basicNack
含义
表示失败确认，一般在消费消息业务异常时用到此方法，可以将消息重新投递入队列。
函数原型
void basicNack(long deliveryTag, boolean multiple, boolean requeue)
- deliveryTag：表示消息投递序号。
- multiple：是否批量确认。
- requeue：值为 true 消息将重新入队列。
basicReject
含义
拒绝消息，与basicNack区别在于不能进行批量操作，其他用法很相似。
函数原型
void basicReject(long deliveryTag, boolean requeue)
- deliveryTag：表示消息投递序号。
- requeue：值为 true 消息将重新入队列。
以上就是详解SpringBoot整合RabbitMQ如何实现消息确认的详细内容，更多关于SpringBoot RabbitMQ消息确认的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要为大家介绍了SpringBoot开发技巧之如何利用AOP记录日志的示例解析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步2021-10-10
java实用型-高并发下RestTemplate的正确使用说明
这篇文章主要介绍了java实用型-高并发下RestTemplate的正确使用说明，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-10-10
springboot整合mybatis实现简单的一对多级联查询功能
这篇文章主要介绍了springboot整合mybatis实现简单的一对多级联查询功能,分步骤通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2021-08-08
文件上传和下载是java web中常见的操作,本文主要介绍了Java批量操作文件系统的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧2024-03-03
这篇文章主要为大家详细介绍了用java实现扫雷游戏，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2022-06-06
在Java开发中,我们经常会面临对List中对象属性去重的需求,本文主要介绍了Java List按照某字段去重的使用示例,具有一定的参考价值,感兴趣的可以了解一下2023-12-12
这篇文章主要介绍了mybatis中的if test判断入参的值问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2022-06-06
这篇文章主要介绍了从dubbo源码分析qos-server端口冲突问题及解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2023-02-02
这篇文章主要介绍了SpringBoot服务访问路径动态处理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2023-12-12
这篇文章主要介绍了Java中获取当前路径的几种方法总结的相关资料,需要的朋友可以参考下2017-02-02
最新评论