一、消费端消息可靠性保证：
-
消息确认（Acknowledgements）：
消费者在接收到消息后，默认情况下RabbitMQ会自动确认消息（autoAck=true）。为保证消息可靠性，可以设置autoAck=false，使得消费者在处理完消息后手动发送确认（basicAck）。如果消费者在处理过程中发生异常或者未完成处理就终止运行，那么消息在超时时间内将不会被删除，会再次被RabbitMQ投递给其他消费者。
2.死信队列（Dead Letter Queue）：
当消息不能被正常消费时（比如达到最大重试次数），可以通过设置TTL（Time To Live）或者死信交换器（Dead Letter Exchange）将消息路由至死信队列，从而有机会后续分析和处理这些无法正常消费的消息。
二、生产端消息可靠性保证：
-
消息持久化：
当生产者发布消息时,可以选择将其标记为持久化（persistent).这意味着即使 RabbitMQ 服务器重启，消息也不会丢失，因为它们会被存储在磁盘上。
2.确认（Confirm）机制：
开启confirm回调模式后，RabbitMQ会在消息成功写入到磁盘并至少被一个交换器接受后，向生产者发送一个确认（acknowledgement）。若消息丢失或无法投递给任何队列，RabbitMQ将会发送一个否定确认（nack). 生产者可以根据这些确认信号判断消息是否成功送达并采取相应的重试策略。
RabbitMQ作为消息中间件并启用publisher confirms（发布者确认）与publisher returns（发布者退回）机制时，可以确保消息从生产者到交换机的投递过程得到更准确的状态反馈。
1.@PostConstruct注解
@PostConstruct注解是Java EE规范中的一部分，主要用于标记在一个Bean初始化完成后需要执行的方法。这个注解由JSR-250定义，并且在Spring框架以及其他遵循Java EE标准的应用服务器中广泛支持。
功能与用途：初始化方法,当容器完成对Bean的实例化并且所有依赖注入完成后，将会自动调用标有@PostConstruct注解的方法。这为开发者提供了一个机会，在对象正式投入使用之前进行一些必要的初始化工作，比如初始化资源、预计算某些值、启动后台任务等增强。
2. Publisher Confirms（发布者确认）
作用： Publisher Confirm机制允许RabbitMQ服务器通知生产者一个消息是否已经被交换机正确接收。当publisher-confirm-type设置为CORRELATED时，RabbitMQ会向生产者发送确认或否定响应，确认消息已到达交换机，但不保证消息已被路由到至少一个队列中。
生产者到交换机的确认（消息到达交换机）
2.1.配置：
spring.rabbitmq.publisher-confirm-type = CORRELATED2.2. 代码实现
只要到达交换机就会触发
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
// 消息成功投递成功并被确认
} else {
// 消息未能正确投递
}
}
});
3.Publisher Returns（发布者退回）
作用： Publisher Return机制用于当消息无法按照路由键规则路由到任何队列时，或者由于其他原因（例如队列满、消息过大等）而被交换机拒绝时，RabbitMQ将消息返回给生产者。
交换机到队列的确认（消息是否正常发送到了队列）
通过实现 ReturnCallback 接口，发送消息失败返回，比如交换机路由不到队列时触发回调：
1.只有消息没有路由到队列的时候，才触发该回调 .文章来源：https://www.toymoban.com/news/detail-851017.html
2.只要有一个队列接受到消息了，它就认为成功.文章来源地址https://www.toymoban.com/news/detail-851017.html
3.1 配置
spring.rabbitmq.publisher-returns = true3.2 代码实现
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
// 处理未被路由或因某种原因被退回的消息
}
});
4.完整代码
4.1消费者
/*
* Copyright (c) 2020, 2024, All rights reserved.
*
*/
package com.by.consumer;
import cn.hutool.core.map.MapUtil;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* <p>Project: spring-boot-rabbitMQ - DirectConsumer</p>
* <p>Powered by scl On 2024-04-07 16:57:20</p>
* <p>描述：<p>
*
* @author 孙臣龙 [1846080280@qq.com]
* @version 1.0
* @since 17
*/
@Configuration
public class ReliabilityConsumer2 {
//注册队列
@Bean
public Queue queue1() {
return QueueBuilder.durable("Re_Q01").deadLetterExchange("dead_E01").deadLetterRoutingKey("DK01").build();
}
//注册交换机
@Bean
public CustomExchange exchange() {
Map<String, Object> map = MapUtil.of("x-delayed-type", "direct");
return new CustomExchange("Re_E01", "x-delayed-message", true, false, map);
}
//绑定交换机和队列
@Bean
public Binding binding2() {
return BindingBuilder.bind(queue1()).to(exchange()).with("RK01").noargs();
}
//注册一个死信交换机
@Bean
public DirectExchange deadExchange() {
return new DirectExchange("dead_E01");
}
//注册一个死信队列
@Bean
public Queue deadQueue() {
return QueueBuilder.durable("dead_Q01").build();
}
//绑定死信交换机和死信队列
@Bean
public Binding deadBinding() {
return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("DK01");
}
//启动一个消费者
@RabbitListener(queues = "Re_Q01")
public void receiveMessage(OrderKO msg) {
System.out.println("消费者2：" + msg);
}
}4.2生产者
/*
* Copyright (c) 2020, 2024, All rights reserved.
*
*/
package com.by.provider;
import com.by.consumer.OrderKO;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.UUID;
/**
* <p>Project: spring-boot-rabbitMQ - DirectProvider</p>
* <p>Powered by scl On 2024-04-07 17:06:41</p>
* <p>描述：<p>
*
* @author 孙臣龙 [1846080280@qq.com]
* @version 1.0
* @since 17
*/
@Service
public class ReliabilityProvider implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}
//启动一个生产者
public void send(OrderKO orderKO) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
System.out.println("callbackSender UUID: " + correlationData.getId());
rabbitTemplate.convertAndSend(
"Re_E01",
"RK01",orderKO,
m-> m,correlationData);
}
@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
if (b) {
System.out.println("消息发送成功");
} else {
System.out.println("消息发送失败");
}
}
@Override
public void returnedMessage(Message message, int i, String s, String s1, String s2) {
System.out.println("消息丢失");
}
}4.3配置文件
spring.rabbitmq.publisher-confirm-type = CORRELATED
spring.rabbitmq.publisher-returns = true4.4测试
@Test
void test6() throws InterruptedException, IOException {
for (int i = 1; i <= 5; i++) {
OrderKO orderKO = OrderKO.builder().id(i).name("孙臣龙" + i).build();
System.out.println("发送消息"+i);
reliabilityProvider.send(orderKO);
}
Thread.sleep(10000);
//System.in.read();
}到了这里，关于Rabbitmq怎么保证消息的可靠性?的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
