RabbitMQ------发布确认高级（八）

可能由于某些意外情况，导致RabbitMQ重启，在RabbitMQ重启过程中，生产者投递消息失败，导致消息丢失。
如果才能够保证RabbitMQ的消息可靠性呢？
可能出现两种问题的情况。
1.交换机接收不到请求
2.队列接收不到请求
解决逻辑：当交换机确认发送消息后，将消息从内存中删除，未能删除的消息，通过定时任务，再次发送给交换机

代码

配置类：

/*** 配置类 发布确认 高级*/
@Configuration
public class ConfirmConfig {//交换机public static final  String CONFIRM_EXCHANGE_NAME = "confirm_exchange";//队列public static final String CONFIRM_QUEUE_NAME = "confirm_queue";//Routingkeypublic static final String CONFIRM_ROUTING_KEY = "key1";//声明交换机@Bean("confirmExchange")public DirectExchange confirmExchange(){return  new DirectExchange(CONFIRM_EXCHANGE_NAME);}//声明队列@Bean("confirmQueue")public Queue confirmQueue(){//创建队列的两种方式//QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();return  new Queue(CONFIRM_QUEUE_NAME);}//绑定@Beanpublic Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue ,@Qualifier("confirmExchange") DirectExchange confirmExchange){//将队列与交换机绑定return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);} }

生产者：

package com.example.springboot01.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;/*** 发送延迟* 生产者*/
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate;//开始发消息  发布确认@GetMapping("/sendConfirmMsg/{message}")public void sendConfirmMsg(@PathVariable String message){log.info("当前时间:{},发送一条消息队列：{}",new Date().toString(),message);/*** 交换机* routingkey* message*/rabbitTemplate.convertAndSend("confirm_exchange","key1",message);}
}

消费者：

package com.example.springboot01.consumer;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Date;/***  消费者*/
@Slf4j
@Component
public class ConfrimConsumer {//接收消息@RabbitListener(queues = "confirm_queue")public void receiveConfirm(Message message) throws Exception {String string = new String(message.getBody(),"UTF-8");log.info("当前时间:{},收到确认消息：{}",new Date().toString(),string);}
}

经测试以上模板代码可行。

交换机收不到消息，实现回调接口

使用回调接口，前提需要在配置文件中增加配置，spring.rabbitmq.publisher-confirm-type有三种参数设置：
1.NONE 禁用发布确认模式，是默认值
2.correlated，发布消息成功到交换机后会触发回调方法（交换机接收成功或失败都会回调）
3.simple
它有两种效果，第一种和correlated一样触发回调方法
第二种：消息在发送成功后使用rabbitTemplete调用waitForConfirms或者waitForConfirmOrDie方法，等待broker节点返回发送结果，根据返回结果判定下一步的逻辑。

注意：waitForConfirmOrDie方法如果返回false，则会关闭channel，则接下来无法发送消息到broker。

spring.rabbitmq.host=182.92.234.71
spring.rabbitmq.prot=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
spring.rabbitmq.publisher-confirm-type=correlated

将发不出去的消息保存下来等等自定义的处理逻辑，都可以通过回调接口自己实现。
在RabbitTemplate中提供了这样的回调接口，

这个接口是一个函数式接口，接口中只有一个方法，可以用lamdba表达式

如果失败了收不到信息，var2就会为false，并且var将失败原因返回。var1是发送的内容
如果成功了，var2将会为true。
实现回调接口，需要创建新的类，对其实现

import lombok.extern.slf4j.Slf4j;
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.Component;import javax.annotation.PostConstruct;@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {//实现类虽然实例化,但是它集成的是内部接口,并且MyCallBack并不在RabbitTemplate里面//所以导致RabbitTemplate在掉用自身接口时,根本掉不到MyCallBack//因此需要将MyCallBack注入到RabbitTemplate@Autowired private RabbitTemplate rabbitTemplate;//注入//PostConstruct注解会在Component、Autowired注解完成后再执行@PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(this);}/*** 交换机确认回调方法    成功和失败都会回调* @param correlationData 保存消息的id以及相关信息* @param ack  交换机收到为true   失败为false* @param cause  成功为null   失败为错误原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("交换机已收到消息,ID为{}的消息",correlationData == null ? "":correlationData.getId());}else {log.info("交换机未收到消息,ID为{}的消息,原因为:",correlationData.getId(),cause);}}
}

correlationData消息的相关信息，本质上不是交换机的，是由生产者发送的。
而生产者的发送消息方法：convertAndSend，有很多重载方法。其中就包括传递correlationData的方法。因此需要再生产发送消息时new出该对象

    public void convertAndSend(String exchange, String routingKey, Object object, @Nullable CorrelationData correlationData) throws AmqpException {this.send(exchange, routingKey, this.convertMessageIfNecessary(object), correlationData);}

生产者代码：

 @GetMapping("/sendConfirmMsg/{message}")public void sendConfirmMsg(@PathVariable String message){log.info("当前时间:{},发送一条消息队列：{}",new Date().toString(),message);/*** 交换机* routingkey* message*/CorrelationData correlationData = new CorrelationData("1");rabbitTemplate.convertAndSend("confirm_exchange","key1",message,correlationData);}

模拟交换机接受不到消息，可以将交换机的名字改错，
结论：将生产者发送消息的队列名改错，就能够再控制台看到，接收失败时的回调函数处理方式。
但是这种，方式只能解决交换机接收不到消息的情况。

队列收不到消息，回退消息

Mandatory参数

在仅开启生产者确认机制的情况下，交换机接收到消息后，会直接将消息生产发送确认消息，如果发现发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃的事件的。
通过设置mandatory参数可以当消息传递过程中不可达目的时将消息返回给生产者。
同样也有退回消息的配置。

spring.rabbitmq.publisher-returns=true

开启配置后，修改config代码，实现回退接口ReturnCallback
同时需要注入

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {//实现类虽然实例化,但是它集成的是内部接口,并且MyCallBack并不在RabbitTemplate里面//所以导致RabbitTemplate在掉用自身接口时,根本掉不到MyCallBack//因此需要将MyCallBack注入到RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;//注入//PostConstruct注解会在Component、Autowired注解完成后再执行@PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(this);//注入rabbitTemplate.setReturnCallback(this);}/*** 交换机确认回调方法    成功和失败都会回调* @param correlationData 保存消息的id以及相关信息* @param ack  交换机收到为true   失败为false* @param cause  成功为null   失败为错误原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info("交换机已收到消息,ID为{}的消息",correlationData == null ? "":correlationData.getId());}else {log.info("交换机未收到消息,ID为{}的消息,原因为:",correlationData.getId(),cause);}}/*** 队列回退接口  当消息不能到目的地时会返回给生产者* 只有不可达目的地时,才进行回退* @param message   消息主体* @param replyCode  退回码值* @param replyText  退回原因* @param exchange   交换机* @param routingKey  路由key*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.error("消息{},被交换机{}退回,退回原因{},路由key{}",new String(message.getBody()),exchange,replyText,routingKey);}
}

修改生产者代码，将routingkey改为key2，队列接收不到，触发回退接口，打印出日志。

备份交换机

当交换机接收到一条不可路由的消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为Fanout，这样就能把所有消息都投递到预期绑定的队列。当然也可以建立一个报警队列，用独立的消费者来进行监测和报警。
示例代码：消费者，主交换机，主队列，备份交换机，备份队列，报警队列，
修改配置文件
1.声明备份交换机，fanout类型
2.声明备份队列
3.声明报警队列
4.备份队列与备份交换机绑定
5.报警队列与备份交换机绑定
6.修改主交换机，配置参数，主交换机发送消息失败后，将消息转发到备份交换机，并且需要设置持久化
ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
.withArgument(“alternate-exchange”,BACKUP_EXCHANGE_NAME).build();

/*** 配置类 发布确认 高级*/
@Configuration
public class ConfirmConfig {//交换机public static final  String CONFIRM_EXCHANGE_NAME = "confirm_exchange";//队列public static final String CONFIRM_QUEUE_NAME = "confirm_queue";//Routingkeypublic static final String CONFIRM_ROUTING_KEY = "key1";//备份交换机public static final  String BACKUP_EXCHANGE_NAME = "backup_exchange";//备份队列public static final String BACKUP_QUEUE_NAME = "backup_queue";//报警队列public static final String WARING_QUEUE_NAME = "warning_queue";//声明交换机@Bean("confirmExchange")public DirectExchange confirmExchange(){return  ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true).withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();}//声明队列@Bean("confirmQueue")public Queue confirmQueue(){//创建队列的两种方式//QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();return  new Queue(CONFIRM_QUEUE_NAME);}//绑定@Beanpublic Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue ,@Qualifier("confirmExchange") DirectExchange confirmExchange){//将队列与交换机绑定return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);}//声明交换机@Bean("backupExchange")public FanoutExchange backupExchange(){return  new FanoutExchange(BACKUP_EXCHANGE_NAME);}//声明备份队列@Bean("backupQueue")public Queue backupQueue(){//创建队列的两种方式//QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();return  new Queue(BACKUP_QUEUE_NAME);}//声明报警队列@Bean("warningQueue")public Queue warningQueue(){//创建队列的两种方式//QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();return  new Queue(WARING_QUEUE_NAME);}//绑定  扇出类型不写routingkey@Beanpublic Binding backupQueueBindingExchange(@Qualifier("backupQueue") Queue backupQueue ,@Qualifier("backupExchange") FanoutExchange backupExchange){//将队列与交换机绑定return BindingBuilder.bind(backupQueue).to(backupExchange);}//绑定  扇出类型不写routingkey@Beanpublic Binding warningQueueBindingExchange(@Qualifier("warningQueue") Queue warningQueue ,@Qualifier("backupExchange") FanoutExchange backupExchange){//将队列与交换机绑定return BindingBuilder.bind(warningQueue).to(backupExchange);}}

消费者

/*** 队列TTL 消费者*/
@Slf4j
@Component
public class DeadLetterConsumer {//接收消息@RabbitListener(queues = "QD")public void receiveD(Message message, Channel channel) throws Exception {String string = new String(message.getBody(),"UTF-8");log.info("当前时间:{},收到死信队列的消息：{}",new Date().toString(),string);}//接收消息@RabbitListener(queues = "confirm_queue")public void receiveConfirm(Message message) throws Exception {String string = new String(message.getBody(),"UTF-8");log.info("当前时间:{},收到确认消息：{}",new Date().toString(),string);}//接收消息  报警消息@RabbitListener(queues = "warning_queue")public void Confirm(Message message) throws Exception {String string = new String(message.getBody(),"UTF-8");log.info("当前时间:{},收到报警消息：{}",new Date().toString(),string);}
}

mandatory参数与备份交换机可以一起使用，如果两者同时开启，消息会去哪里？
备份交换机的优先级高！