rabbitmq可靠性投递_Rabbitmq的使用五_消息的可靠性投递

Rabbitmq的使用五_消息的可靠性投递

官网地址&＃xff1a;https://www.rabbitmq.com/tutorials/tutorial-seven-java.html

通过前面的学习&＃xff0c;可以知道消息的可靠性投递&＃xff0c;可以做持久化操作

过程一&＃xff1a;从生产者发送消息到RabbitMQ服务器的过程。

过程二&＃xff1a;确保消息从交换机路由到队列

过程三&＃xff1a;确保消息在队列中正确的存储

过程四&＃xff1a;确保消息从队列正确地投递到消费者

消息本身也要做持久化操作。

2.Rabbitmq的工作模型

执行流程&＃xff1a;生产者发送消息通过channel和rabbitmq建立连接&＃xff0c;发送一条携带路由关键字的消息到交换机上面的时候。交换机根据消息携带关键字&＃xff0c;去查哪一些队列绑定了这些关键字&＃xff0c;然后把消息路由到这些队列上面&＃xff0c;然后消费者从队列中取消息就可以了

一个rabbitmq里面&＃xff0c;可以建立多个不同的交换机和绑定队列&＃xff0c;这时就用到我们的virtual host虚拟机。我们可以把每一个虚拟机当做一个rabbitmq的服务器。这样做解决了硬件资源的问题。因此&＃xff0c;我们可以创建多个虚拟机&＃xff0c;每一个虚拟机都可以看做是一个rabbitmq的服务器&＃xff0c;可以根据此来创建很多的交换机和队列&＃xff0c;以及定义之间的绑定关系&＃xff0c;然后我们去创建一些用户&＃xff0c;比如我们的资金系统&＃xff0c;有资金系统的虚拟机&＃xff0c;我们把资金系统的用户分配资金系统的权限就ok了。那么对应的其他系统&＃xff0c;每一个系统都有自己的用户&＃xff0c;自己的权限&＃xff0c;那么不同的虚拟机之间是完全透明的。他们之间可以建立同名的交换机&＃xff0c;通过这样的方式&＃xff0c;我们可以实现硬件资源的高效利用和硬件资源的隔离。

生产者或者消费者和rabbitmq之间的连接是一个长连接。如果我们直接连接&＃xff0c;频繁的创建连接&＃xff0c;就会造成性能问题&＃xff0c;我们引入一个channel的通道。这是一些虚拟的连接。我们需要连接rabbitmq的话&＃xff0c;直接从这些虚拟的连接中拿一个连接就可以使用了。

exchange&＃xff1a;本质是地址的清单,本身不存储消息。相当于一个路由功能

引入交换机的作用&＃xff1a;是为了达到消息的一个灵活的投递

能够创建多少队列&＃xff1a;这个主要取决于队列是保存在哪里的&＃xff0c;如果队列是保存在内存里的&＃xff0c;那么创建多少队列&＃xff0c;取决于内存空间的大小&＃xff0c;如果队列是保存在硬盘上的&＃xff0c;取决于硬盘空间的大小

如何实现消息的灵活投递&＃xff1a;交换机和队列之间的关系是多对多的关系

2.1如果过程一发生了异常怎么办&＃xff1f;

从生产者发送消息到rabbitmq服务器的过程中&＃xff0c;失败了。RabbitMQ给我们提供了两种方式&＃xff1a;1.AMQP事务  2.Confirm模式

1.事务模式&＃xff1a;每一个事务都要等待消息的一个应答&＃xff0c;所以事务模式极其消耗性能的一个东西。因为极其消耗性能&＃xff0c;所以在生产中一般也不会使用&＃xff0c;因为&＃xff0c;如果每一条消息&＃xff0c;都开启事务&＃xff0c;太消耗性能了。

事务模式的三个方法&＃xff1a;channel.txSelect(); channel.txCommit(); channel.txRollback();

代码案例&＃xff1a;

public classRabbitMqTransActionSender {//事务队列

private static final String TRANSTATION_QUEUE &＃61; "transaction_queue";public static void main(String[] args) throwsIOException, TimeoutException {//1.获取连接

Connection connection &＃61;RabbitMQConnectionFactory.getRabbitMQConnections();//2.获取通道

Channel channel &＃61;connection.createChannel();//3.声明队列

channel.queueDeclare(TRANSTATION_QUEUE, false, false, false, null);//4.发送消息

String msg &＃61; "小河流水哗啦啦,我和姐姐去采花1";//将channel设置为事务模式

try{

channel.txSelect();

channel.basicPublish("", TRANSTATION_QUEUE, null, msg.getBytes());int a &＃61; 10 / 0;

channel.txCommit();

System.out.println("消息已经提交");

}catch(Exception e) {

channel.txRollback();

System.out.println("消息已经回滚");

}

channel.close();

connection.close();

}

}

View Code

演示正常情况&＃xff0c;发送者发送消息&＃xff0c;消费者正常接收消息

消息发送过程中&＃xff0c;出现异常。消息回滚掉&＃xff0c;消费者不会接收到任何消息

图4生产者　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　图5消费者

2.确认模式

当我们发送消息成功之后&＃xff0c;会有一个ack应答&＃xff0c;只要我们的channel.waitForConfirms返回一个true,表示我们的消息就是发送成功的。

rabbitmq的消息确认&＃xff0c;默认不启动了&＃xff0c;需要开启

Channel channel &＃61;connection.createChannel();

channel.confirmSelect();

单条发送消息

当发送者发送消息报错时&＃xff0c;消费者就不会受到消息

public classRabbitMqConfirmSender {private static final String CONFIRM_QUEUE &＃61; "confirm_queue";public static void main(String[] args) throwsIOException, TimeoutException, InterruptedException {//1.获取连接

Connection connection &＃61;RabbitMQConnectionFactory.getRabbitMQConnections();//2.获取通道

Channel channel &＃61;connection.createChannel();//3.声明队列

channel.queueDeclare(CONFIRM_QUEUE, false, false, false, null);//4.发送消息

String msg &＃61; "小河流水哗啦啦,我和姐姐去采花1";//将channel设置为confirm模式

channel.confirmSelect();

channel.basicPublish("", CONFIRM_QUEUE, null, msg.getBytes());if(channel.waitForConfirms()) {

System.out.println("消息发送成功");

}

channel.close();

connection.close();

}

}

View Code

我们在开始时提到&＃xff0c;代理异步确认已发布的消息&＃xff0c;代码会同步等待&＃xff0c;直到消息确认为止。客户端实际异步接收确认&＃xff0c;并解除调用阻塞&＃xff0c;可以看作是一个在内部依赖于异步通知的同步助手。

单条消息的发送确认模式&＃xff0c;效率也是很低的&＃xff0c;每一条消息&＃xff0c;先开启消息确认机制&＃xff0c;然后发送&＃xff0c;然后处理消息应答&＃xff0c;效率太低。

批量发送消息

public classRabbitMqConfirmBatchSender {private static final String BATCH_CONFIRM_QUEUE &＃61; "batch_confirm_queue";public static void main(String[] args) throwsIOException, TimeoutException, InterruptedException {//1.获取连接

Connection connection &＃61;RabbitMQConnectionFactory.getRabbitMQConnections();//2.获取通道

Channel channel &＃61;connection.createChannel();//3.声明队列

channel.queueDeclare(BATCH_CONFIRM_QUEUE, false, false, false, null);//4.发送消息//将channel设置为confirm模式

channel.confirmSelect();

String msg&＃61; "小河流水哗啦啦,我和姐姐去采花";for (int i &＃61; 1; i <&＃61; 100; i&＃43;&＃43;) {//int a &＃61; 19/0;

channel.basicPublish("", BATCH_CONFIRM_QUEUE, null, (msg &＃43;i).getBytes());

}if(channel.waitForConfirms()) {

System.out.println("消息发送成功");

}

channel.close();

connection.close();

}

}

View Code

当批量发送异常的时候&＃xff0c;消费者不会收到任何消息,如下图所示&＃xff1a;

这种批量发送消息的方式&＃xff0c;只要有一条消息未被broker确认&＃xff0c;就会发生异常,也就是说当我们channel.waitFroConfirms;只要不抛出异常&＃xff0c;就可以认为我们的消息发送成功了

但是:这种方式也有缺点&＃xff1a;

1.我们是积累多少条消息进行消息的发送

2.我们假设是1000条发送消息一次&＃xff0c;如果前999条发送失败&＃xff0c;刚好第1000条发送失败了&＃xff0c;怎么办

Rabbitmq官网还给我们提供了一种方式&＃xff0c;是采用异步的方式&＃xff0c;进行消息的收发&＃xff0c;使用异步的方式&＃xff0c;是可以进行一边发送&＃xff0c;一边确认的方式&＃xff0c;进行消息的收发的&＃xff0c;异步的情况&＃xff0c;消息不会自动重发的

Broker异步确认&＃xff1a;只需要在发送者客户端注册一个异步回调&＃xff0c;就可以接收到确认消息

代码如下:

public classRabbitMqAsyncConfirmSender {private static final String ASYNC_CONFIRM_QUEUE &＃61; "async_confirm_queue";public static void main(String[] args) throwsIOException, TimeoutException, InterruptedException {//1.获取连接

Connection connection &＃61;RabbitMQConnectionFactory.getRabbitMQConnections();//2.获取通道

Channel channel &＃61;connection.createChannel();//3.声明队列

channel.queueDeclare(ASYNC_CONFIRM_QUEUE, false, false, false, null);//4.发送消息

String msg &＃61; "小河流水哗啦啦,我和姐姐去采花";//将channel设置为confirm模式

channel.confirmSelect();//发送消息

for (int i &＃61; 1; i <&＃61; 20; i&＃43;&＃43;) {//if (i&＃61;&＃61;17){//int a &＃61; 10/0;//}

channel.basicPublish("", ASYNC_CONFIRM_QUEUE, null, (msg &＃43; "&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;>" &＃43;i).getBytes());

}

channel.addConfirmListener(newConfirmListener() {

&＃64;Overridepublic void handleAck(long deliveryTag, boolean multiple) throwsIOException {

System.out.print("已确认的消息,标识&＃xff1a;" &＃43;deliveryTag);

System.out.println("多个消息&＃xff1a; " &＃43;multiple);

}

&＃64;Overridepublic void handleNack(long deliveryTag, boolean multiple) throwsIOException {

System.out.println("Broker 未确认消息,标识: " &＃43;deliveryTag);

}

});

System.out.println("程序执行完成");

}

}

View Code

执行结果如下&＃xff1a;

在某些应用程序中&＃xff0c;确保已发布的消息到达broker可能非常重要。发布者确认是RabbitMQ的一个特性&＃xff0c;有助于满足这一需求。发布者确认在本质上是异步的&＃xff0c;但是也可以同步地处理它们。

1.单条发布消息&＃xff0c;同步等待确认:简单&＃xff0c;但吞吐量非常有限。

2.批量发布消息&＃xff0c;同步等待批处理的确认:简单、合理的吞吐量&＃xff0c;但是很难判断什么时候出现了错误

3.异步处理:最佳的性能和资源的使用&＃xff0c;在错误的情况下良好的控制。

2.2 过程二发送失败了怎么办?

就是交换机路由消息到队列的过程中发送失败了。

解决办法&＃xff1a;可以给当前交换机设置备份交换机

交换机本身也是支持持久化操作的。在声明交换机的时候&＃xff0c;参数三表示是否持久化

/*** Actively declare a non-autodelete exchange with no extra arguments

*&＃64;seecom.rabbitmq.client.AMQP.Exchange.Declare

*&＃64;seecom.rabbitmq.client.AMQP.Exchange.DeclareOk

*&＃64;paramexchange the name of the exchange

*&＃64;paramtype the exchange type

*&＃64;paramdurable true if we are declaring a durable exchange (the exchange will survive a server restart)

*&＃64;throwsjava.io.IOException if an error is encountered

*&＃64;returna declaration-confirm method to indicate the exchange was successfully declared*/Exchange.DeclareOk exchangeDeclare(String exchange, String type,boolean durable) throws IOException;

2.3 过程三失败了怎么办&＃xff1f;

当我们没有对消息进行配置的时候&＃xff0c;默认是保存在内存中的&＃xff0c;消息保存在内存中&＃xff0c;不可避免的会出现rabbitmq服务器的重启&＃xff0c;宕机等问题&＃xff0c;所以&＃xff0c;可以对消息做一个消息的持久化的处理。

发送消息的时候&＃xff0c;可以给消息设置一个properties的属性&＃xff0c;通过配置properties可以配置消息的一个持久化操作。

队列可以做持久化操作、消息也可以做持久化操作。

2.4过程四失败了怎么办&＃xff1f;

确保消息从队列正确地投递到消费者ack应答

消费者接收到消息的时候&＃xff0c;就会给broker一个应答&＃xff0c;broker拿到应答之后&＃xff0c;就会从队列中删除这条消息。而不是在方法执行完之后&＃xff0c;再给服务器的应答。我们可以手动的执行一个ack应答机制。

2.5 如果消费者处理消息的时候&＃xff0c;抛出异常了&＃xff0c;生产者怎么知道&＃xff1f;

1.消费者回调

1) 发送回执(表示每发送一条消息&＃xff0c;都给生产者一条消息回执)

2)生产者提供api(生产者发送消息的时候&＃xff0c;保存一条消息入库)&＃xff0c;消费者调用暴露的api&＃xff0c;去修改这条消息的状态。如果我们消息的状态没有发生变更的话&＃xff0c;那么我就可以判断有可能消费者在处理消息的时候&＃xff0c;发生了问题

2.补偿机制

如果没有发送回执&＃xff1a;可能是因为网络问题

如果没有调用api:可能消费者调用生产者的过程中出现了问题

在以上的情况都没有使用的情况下面&＃xff0c;我们就需要使用一个补偿机制&＃xff0c;比如

1)消息的重发&＃xff1a;

重发的前提:我们发送前需要把数据保存到数据库中&＃xff0c;然后重发的时候&＃xff0c;直接从数据库中进行消息的获取&＃xff0c;然后重新进行一个消息的发送。但是如果发送5次&＃xff0c;10次&＃xff0c;如果消费者一直没有应答。。。就会一直重发。所以这里一定要做一个次数的控制&＃xff0c;等达到一定的次数之后&＃xff0c;我们就不进行重发操作了&＃xff0c;我们会在夜间进行一个对账的操作。

可能发送消息的时候&＃xff0c;就是消费成功了&＃xff0c;但是由于网络原因&＃xff0c;回执执行慢了&＃xff0c;

3.消息的幂等性

1)处理1次消息&＃xff0c;跟处理10 次消息的结果都是一样的

2) 消息必须有一个唯一性的标志&＃xff1a;在金融系统中&＃xff0c;任何一笔交易&＃xff0c;都会有一个全局流水号的标志, message发送消息的时候&＃xff0c;有一个消息的id&＃xff0c;消息id&＃43;业务id唯一判断 (重帐控制)消费会不会重复消费

3.如果保证消息的顺序消费

多个生产者和多个消费者的情况&＃xff0c;基本是没有办法实现一个消息的顺序消费的&＃xff0c;比如发送了3条消息&＃xff0c;消费者的消费速率是不一样&＃xff0c;所以我们无法保证哪一个消息是先消费完成的。所以我们完成消费的顺序性&＃xff0c;如果我们只有一个生产者&＃xff0c;和一个消费者&＃xff0c;根据队列先进先出的思想&＃xff0c;我们是可以保证消息的一个顺序消费的。对于每一组消息&＃xff0c;我们都有一个parentID(批次号),也有一个seqNo&＃xff0c;如果上一个批次的消息没有消费完&＃xff0c;就不能消费下一个消息的

int sequenceNumber &＃61;channel.getNextPublishSeqNo());

ch.basicPublish(exchange, queue, properties, body);

当然如果要保证消息的一个顺序性的消费&＃xff0c;我们可以使用rocketmq的顺序消费&＃xff0c;他本身就是支持的

4.springboot整合rabbitmq实现事务模式

在SpringBoot项目中&＃xff0c;使用RabbitMQ事务&＃xff0c;只需要声明一个事务管理的Bean&＃xff0c;并将RabbitTemplate的事务设置为true即可

消费者方

1.消费者添加配置文件

server:

port: 8010

spring:

rabbitmq:

host: ip地址

username: yingxiaocao

password: yingxiaocao

virtual-host: /yingxiaocao

listener:

type: simple

simple: # 开启手动应答

acknowledge-mode: manual

配置mq地址

2.声明队列、交换、绑定关系、事务bean(消费者和生产者都要配置。否则当生产者启动&＃xff0c;消费者未启动时&＃xff0c;发送消息&＃xff0c;就会报错)

&＃64;PropertySource("classpath:transaction_mq.properties")

&＃64;Configurationpublic classConsumerConfig {

&＃64;Value("${TRANSACTION_EXCHANGE_NAME}")privateString exchangeName;

&＃64;Value("${FIRST_QUEUE}")privateString firstQueue;//1.声明一个交换机

&＃64;Bean("fanout_exchange")publicFanoutExchange getFanoutExchange() {

FanoutExchange fanoutExchange&＃61; newFanoutExchange(exchangeName);returnfanoutExchange;

}//2.声明2个队列

&＃64;Bean("first_queue")publicQueue getFirstQueue() {

Queue queue&＃61; newQueue(firstQueue);returnqueue;

}//3.绑定关系

&＃64;Beanpublic Binding bindingExchange(&＃64;Qualifier("fanout_exchange") FanoutExchange fanoutExchange, &＃64;Qualifier("first_queue") Queue queue) {returnBindingBuilder.bind(queue).to(fanoutExchange);

}

&＃64;BeanpublicRabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory cachingConnectionFactory) {return newRabbitTransactionManager(cachingConnectionFactory);

}

声明队列、交换机

3.创建一个消费者

&＃64;Component

&＃64;PropertySource("classpath:transaction_mq.properties")public classRabbitmqConsumer {

&＃64;RabbitListener(queues&＃61; {"${FIRST_QUEUE}"})public void receive(Message message, Channel channel) throwsIOException {

String msg&＃61; newString(message.getBody());

System.out.println("接收到的消息: " &＃43;msg);

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

}

}

消费者

4.创建一个启动类

&＃64;SpringBootApplicationpublic classTransactionConsumerStartApp {public static voidmain(String[] args) {

SpringApplication.run(TransactionConsumerStartApp.class, args);

}

}

消费者启动类

生产者方

1.创建一个controller

&＃64;RequestMapping("/producer")

&＃64;RestControllerpublic classProducerController {

&＃64;AutowiredprivateProducerService producerService;

&＃64;RequestMapping("/send")public voidsend(String msg) {

producerService.send(msg);

}

}

controller

2.创建一个service,用来发送消息

&＃64;PropertySource("classpath:transaction_mq.properties")

&＃64;Servicepublic classProducerService {

&＃64;Value("${TRANSACTION_EXCHANGE_NAME}")privateString exchangeName;

&＃64;AutowiredprivateRabbitTemplate rabbitTemplate;

&＃64;PostConstructpublic voidinit() {//创建对象的同时&＃xff0c;开启channel事务模式

rabbitTemplate.setChannelTransacted(true);

}/*** 发送消息

*&＃64;parammsg*/&＃64;Transactionalpublic voidsend(String msg) {

rabbitTemplate.convertAndSend(exchangeName,"",msg);

System.out.println("消息已发送: "&＃43;msg);if (msg.equals("xxx")) {throw new RuntimeException("抛出异常了");

}

}/*** 配置启用rabbitmq事务

*&＃64;paramconnectionFactory

*&＃64;return

*/&＃64;BeanpublicRabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory connectionFactory) {return newRabbitTransactionManager(connectionFactory);

}

}

service发送消息

3.创建一个启动类

&＃64;SpringBootApplicationpublic classTransactionProducerStartApp {public static voidmain(String[] args) {

SpringApplication.run(TransactionProducerStartApp.class, args);

}

}

View Code

4.声明配置

&＃64;PropertySource("classpath:transaction_mq.properties")

&＃64;Componentpublic classRabbitConfig {

&＃64;Value("${TRANSACTION_EXCHANGE_NAME}")privateString exchangeName;

&＃64;Value("${FIRST_QUEUE}")privateString firstQueue;//1.声明一个交换机

&＃64;Bean("fanout_exchange")publicFanoutExchange getFanoutExchange() {

FanoutExchange fanoutExchange&＃61; newFanoutExchange(exchangeName);returnfanoutExchange;

}//2.声明1个队列

&＃64;Bean("first_queue")publicQueue getFirstQueue() {

Queue queue&＃61; newQueue(firstQueue);returnqueue;

}//3.绑定关系

&＃64;Beanpublic Binding bindingExchange(&＃64;Qualifier("fanout_exchange") FanoutExchange fanoutExchange, &＃64;Qualifier("first_queue") Queue queue) {returnBindingBuilder.bind(queue).to(fanoutExchange);

}

&＃64;BeanpublicRabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory cachingConnectionFactory) {return newRabbitTransactionManager(cachingConnectionFactory);

}

}

声明队列、交换机、绑定关系

执行结果&＃xff1a;

正常情况&＃xff0c;生产者发送消息&＃xff0c;消费者收到消费

异常情况&＃xff1a;生产者发送消息后&＃xff0c;抛出异常了&＃xff0c;消费者并没有收到消息&＃xff0c;可见消息回滚了

但是事务模式效率太差。

5.springboot整合rabbitmq实现消息确认模式

通过生产者确认机制&＃xff0c;生产者可以在消息被服务器成功接收时得到反馈&＃xff0c;并有机会处理未被成功接收的消息。

在Springboot中开启RabbitMQ的生产者确认模式也很简单&＃xff0c;只多了一行配置&＃xff1a;publisher-confirms: true即表示开启生产者确认模式。

server:

port: 9010

spring:

rabbitmq:

username: yingxiaocao

password: yingxiaocao

host: 地址

virtual-host: /yingxiaocao

listener:

type: simple

publisher-confirms: true # 开启消息确认模式

View Code

改变生产者代码

&＃64;PropertySource("classpath:confirm_mq.properties")

&＃64;Service

&＃64;Slf4jpublic class ProducerService implementsRabbitTemplate.ConfirmCallback {

&＃64;Value("${CONFRIM_EXCHANGE_NAME}")privateString exchangeName;

&＃64;AutowiredprivateRabbitTemplate rabbitTemplate;

&＃64;PostConstructpublic voidinit() {//创建对象的同时&＃xff0c;开启channel事务模式//rabbitTemplate.setChannelTransacted(true);//开启确认模式

rabbitTemplate.setConfirmCallback(this);

}/*** 发送消息

*

*&＃64;parammsg*/

public voidsend(String msg) {//创建一个消息编号

CorrelationData correlationData &＃61; newCorrelationData(UUID.randomUUID().toString());

rabbitTemplate.convertAndSend(exchangeName,"aaa", msg, correlationData);

log.info("消息id{},路由key{},发送消息内容{}", correlationData.getId(), "aaa", msg);//创建一个无法投递成功的消息

CorrelationData correlationData1 &＃61; newCorrelationData(UUID.randomUUID().toString());

rabbitTemplate.convertAndSend(exchangeName,"bbb", msg, correlationData1);

log.info("消息id{},路由key{},发送消息内容{}", correlationData1.getId(), "bbb", msg);

}/*** 消息回调

*

*&＃64;paramcorrelationData

*&＃64;paramack

*&＃64;paramcause*/&＃64;Overridepublic void confirm(CorrelationData correlationData, booleanack, String cause) {

String id&＃61; correlationData !&＃61; null ? correlationData.getId() : "";if(ack) {

log.info("消息投递成功,消息id{}", id);

}else{

log.info("消息投递失败,消息id{},原因:{}", id, cause);

}

}

}

View Code

执行代码结果如下&＃xff1a;

生产者截图

消费者截图。

由运行结果可知。消息只要发送到了交换机&＃xff0c;不管消息有没有成功投递到队列里面&＃xff0c;都会给生产者一个ack应答。

如何让消息被路由到队列后再返回ACK呢&＃xff1f;

1.设置mandatory参数

设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

当把 mandotory 参数设置为 true 时&＃xff0c;如果交换机无法将消息进行路由时&＃xff0c;会将该消息返回给生产者&＃xff0c;而如果该参数设置为false&＃xff0c;如果发现消息无法进行路由&＃xff0c;则直接丢弃。

rabbitTemplate.setMandatory(true);

如果要让消息返回给生产者&＃xff0c;需要添加一个回调

为了进行回调&＃xff0c;我们需要实现一个接口 RabbitTemplate.ReturnCallback

使用mandatory这种方式&＃xff0c;如果消息发送失败&＃xff0c;返回给生产者&＃xff0c;通过看日志的方式&＃xff0c;就比较麻烦了。可以使用备份交换机的方式。如果消息从交换机路由到队列失败&＃xff0c;转发给备份交换机&＃xff0c;由备份交换机绑定的队列&＃xff0c;进行处理。就比较好了

2.设置备份交换机

创建交换机的时候&＃xff0c;为交换机添加备份交换机代码&＃xff0c;可为备份交换机添加不同的队列&＃xff0c;实现不同的功能

.withArgument("alternate-exchange",BUSINESS_BACKUP_EXCHANGE_NAME);