目录
一、发布确认SpringBoot版本
确认机制图例：
代码实战：
代码架构图：
1.1交换机的发布确认
添加配置类
消息消费者
消息生产者发布消息后的回调接口
测试：
1.2回退消息并重发（队列的发布确认）
修改回调接口
生产者：
测试：
二、备份交换机
实战
生产者
报警消费者：
测试：
一、发布确认SpringBoot版本
首先发布消息后进行备份在缓存里，如果消息成功发布确认到交换机，则从缓存里删除该消息，如果没有成功发布，则设置一个定时任务，重新从缓存里获取消息发布到交换机，直到成功发布到交换机。
确认机制图例：
代码实战：
一个交换机：confirm.exchange，一个队列：confirm.queue，一个消费者：confirm.consumer
其中交换机类型时 direct，与队列关联的 routingKey 是 key1
代码架构图：
1.1交换机的发布确认
配置文件中添加：
server:
port: 8888
spring:
rabbitmq:
host: 192.168.163.132
port: 5672
username: 2252631565
password: 2252631565
# 高级发布确认 发布消息成功后将会触发回调方法
publisher-confirm-type: correlatedNONE值是禁用发布确认模式，是默认值CORRELATED值是发布消息成功到交换器后会触发回调方法SIMPLE值经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker;
添加配置类
声明交换机和队列，并且将交换机和队列进行绑定:
@Configuration
public class ConfirmConfig {
public static final String CONFIRM_EXCHANGE="confirm.exchange";
public static final String CONFIRM_QUEUE="confirm.queue";
public static final String ROUTING_KEY="key1";
@Bean
public DirectExchange confirmExchange(){
return new DirectExchange(CONFIRM_EXCHANGE,false,false);
}
@Bean
public Queue confirmQueue(){
return QueueBuilder.durable(CONFIRM_QUEUE).build();
}
@Bean
public Binding EAndQBind(@Qualifier("confirmExchange") DirectExchange exchange,@Qualifier("confirmQueue")Queue queue){
return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
}
}消息生产者
也可以说是 Controller 层，在这里发送两条消息给两个交换机，其中一个交换机是我们设置好的，另一个交换机不存在；这样就可以清晰看出交换机应答效果。
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
@Autowired
private RabbitTemplate rabbitTemplate;
//高级发布确认模式
@GetMapping("/sendConfirmMsg/{message}")
public void sendConfirmMsg(@PathVariable String message){
log.info("发送一条时长为的消息给第一个队列内容是:{}",new Date().toString(),message);
CorrelationData correlationData=new CorrelationData("1");
correlationData.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);
//向一个不存在的交换机发送消息
log.info("发送一条时长为的消息给第一个队列内容是:{}",new Date().toString(),message);
CorrelationData correlationData2=new CorrelationData("2");
correlationData2.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE+123,ConfirmConfig.ROUTING_KEY,message,correlationData2);
}
}
消息消费者
监听 confirm.queue 队列
@Slf4j
@Component
public class ConfirmLetterQueue {
@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)
public void confirmConsumer(Message message, Channel channel){
log.info("收到了消息：{}",new String(message.getBody()));
}
}消息生产者发布消息后的回调接口
只要生产者发布消息，交换机不管是否收到消息，都会调用该类的 confirm 方法
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init(){
//注入
rabbitTemplate.setConfirmCallback(this::confirm);
}
/*交换机确认回调
1.交换机收到了消息 触发回调
1.1 correlationData(我们在发消息的时候自己创建的) 消息的ID以及消息内容
1.2 ack 交换机收到消息 true
1.3 cause 交换机收到消息的原因 null
---------------------------------
2.交换机未收到消息 触发回调
2.1 correlationData 消息的ID以及消息内容
2.2 ack 交换机未收到消息 false
2.3 cause 失败原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
log.info("发送消息到交换机成功！消息体为：{}",new String(correlationData.getReturned().getMessage().getBody()));
}else {
log.info("发送消息到交换机失败！原因为：{}",cause.toString());
}
}
}
测试：
效果很明显，我们配置的交换机成功收到消息并转发给队列；不存在的交换机没有接受到消息并作出反应。
1.2回退消息并重发（队列的发布确认）
在配置文件中开启消息回退功能
server:
port: 8888
spring:
rabbitmq:
host: 192.168.163.133
port: 5672
username: 2252631565
password: 2252631565
# 高级发布确认 发布消息成功后将会触发回调方法
publisher-confirm-type: correlated
# 消息回退 当消息未路由至队列时触发
publisher-returns: true修改回调接口
实现 RabbitTemplate.ReturnsCallback 接口，并实现方法
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init(){
//注入
rabbitTemplate.setConfirmCallback(this::confirm);
rabbitTemplate.setReturnsCallback(this::returnedMessage);
}
/*交换机确认回调
1.交换机收到了消息 触发回调
1.1 correlationData(我们在发消息的时候自己创建的) 消息的ID以及消息内容
1.2 ack 交换机收到消息 true
1.3 cause 交换机收到消息的原因 null
---------------------------------
2.交换机未收到消息 触发回调
2.1 correlationData 消息的ID以及消息内容
2.2 ack 交换机未收到消息 false
2.3 cause 失败原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
log.info("时间:{}发送消息到交换机成功！",new Date());
}else {
log.info("发送消息到交换机失败！原因为：{}",cause.toString());
}
}
//当消息未路由到队列时触发 只有失败时才触发 若消息发送至延迟队列则一定会触发回退 记得根据交换机名称排除延迟队列
@Override
public void returnedMessage(ReturnedMessage returned) {
log.error("消息：'{}',被交换机：{}回退，回退原因为：{}，routingKey为：{}"
,new String(returned.getMessage().getBody())
,returned.getExchange()
,returned.getReplyText()
,returned.getRoutingKey());
//10s后消息重发
try {
Thread.sleep(10000);
log.info("时间:{}，生产者重新发消息",new Date());
rabbitTemplate.convertAndSend(returned.getExchange(),ConfirmConfig.ROUTING_KEY,new String(returned.getMessage().getBody()));
}catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
生产者：
向交换机中发送消息，指定错误的routingkey,触发队列回退消息并重发消息。
//高级发布确认模式
@GetMapping("/sendConfirmMsg/{message}")
public void sendConfirmMsg(@PathVariable String message){
//向一个不存在的队列发送消息
log.info("时间:{}生产者发送一条的消息给第一个队列内容是:{}",new Date().toString(),message);
CorrelationData correlationData2=new CorrelationData("2");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY+123,message,correlationData2);
}测试：
回退未进入队列的消息并重新发送消息。
二、备份交换机
什么是备份交换机呢？备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout （扇出），这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进 入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警并可以重发消息。
实战
需要一个备份交换机 backup.exchange，类型为 fanout，该交换机发送消息到队列 backup.queue 和 warning.queue。
修改高级确认发布 配置类
@Configuration
public class ConfirmConfig {
public static final String CONFIRM_EXCHANGE="confirm.exchange";
public static final String CONFIRM_QUEUE="confirm.queue";
public static final String ROUTING_KEY="key1";
//备份交换机
public static final String BACKUP_EXCHANGE="backup.exchange";
//备份队列
public static final String BACKUP_QUEUE="backup.queue";
//报警队列
public static final String WARNING_QUEUE="warning.queue";
@Bean
public DirectExchange confirmExchange(){
//绑定确认交换机与备份交换机
Map<String,Object> argument=new HashMap<>();
argument.put("alternate-exchange",BACKUP_EXCHANGE);
return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).withArguments(argument).build();
}
//备份交换机
@Bean
public FanoutExchange backupExchange(){
return new FanoutExchange(BACKUP_EXCHANGE);
}
@Bean
public Queue confirmQueue(){
return QueueBuilder.durable(CONFIRM_QUEUE).build();
}
//备份队列
@Bean
public Queue backupQueue(){
return QueueBuilder.durable(BACKUP_QUEUE).build();
}
//警告队列
@Bean
public Queue warningQueue(){
return QueueBuilder.durable(WARNING_QUEUE).build();
}
@Bean
public Binding EAndQBind(@Qualifier("confirmExchange") DirectExchange exchange,@Qualifier("confirmQueue")Queue queue){
return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
}
//绑定备份交换机与两个队列
@Bean
public Binding BAndBBing(@Qualifier("backupExchange") FanoutExchange exchange,@Qualifier("backupQueue")Queue queue){
return BindingBuilder.bind(queue).to(exchange);
}
@Bean
public Binding BAndWBing(@Qualifier("backupExchange") FanoutExchange exchange,@Qualifier("warningQueue")Queue queue){
return BindingBuilder.bind(queue).to(exchange);
}
}
生产者
生产者发送两条消息 一个配置正确的路由，另一个是错误的路由。预期目标是正确路由正常接收消息，错误路由传输的信息由警告队列接收。
//高级发布确认模式
@GetMapping("/sendConfirmMsg/{message}")
public void sendConfirmMsg(@PathVariable String message){
log.info("时间:{}生产者发送两条消息队列内容是:{}",new Date().toString(),message);
CorrelationData correlationData=new CorrelationData("1");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);
//向一个不存在的队列发送消息
CorrelationData correlationData2=new CorrelationData("2");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY+123,message,correlationData2);
}报警消费者：
接收不可路由的消息
/**
* 报警消费者
*/
@Slf4j
@Component
public class WarningConsumer {
//监听报警消息
@RabbitListener(queues = ConfirmConfig.WARNING_QUEUE)
public void receiveWarningMsg(Message message){
String msg=new String(message.getBody());
log.info("报警发现不可路由消息：{},重发消息",msg);
}
}
测试：
生产者发送两条消息 一个配置正确的路由，另一个是错误的路由。预期目标是正确路由正常接收消息，错误路由传输的信息由警告队列接收。
在此案例中，也设置了消息回退的配置，但是没有触发消息回退。由此得出：备份交换机的优先级更高。
