网站建设未完成,semester,投融资平台,网站无内容 备案目录 一、发布确认SpringBoot版本
确认机制图例#xff1a;
代码实战#xff1a;
代码架构图#xff1a;
1.1交换机的发布确认
添加配置类
消息消费者
消息生产者发布消息后的回调接口
测试#xff1a; 1.2回退消息并重发#xff08;队列的发布确认#xff09;
…目录 一、发布确认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.132port: 5672username: 2252631565password: 2252631565
# 高级发布确认 发布消息成功后将会触发回调方法publisher-confirm-type: correlated
NONE 值是禁用发布确认模式是默认值CORRELATED 值是发布消息成功到交换器后会触发回调方法SIMPLE 值经测试有两种效果其一效果和 CORRELATED 值一样会触发回调方法其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果根据返回结果来判定下一步的逻辑要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel则接下来无法发送消息到 broker;
添加配置类
声明交换机和队列并且将交换机和队列进行绑定:
Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGEconfirm.exchange;public static final String CONFIRM_QUEUEconfirm.queue;public static final String ROUTING_KEYkey1;Beanpublic DirectExchange confirmExchange(){return new DirectExchange(CONFIRM_EXCHANGE,false,false);}Beanpublic Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE).build();}Beanpublic 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 {Autowiredprivate RabbitTemplate rabbitTemplate;//高级发布确认模式GetMapping(/sendConfirmMsg/{message})public void sendConfirmMsg(PathVariable String message){log.info(发送一条时长为的消息给第一个队列内容是:{},new Date().toString(),message);CorrelationData correlationDatanew 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 correlationData2new CorrelationData(2);correlationData2.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE123,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 {Autowiredprivate RabbitTemplate rabbitTemplate;PostConstructpublic void init(){//注入rabbitTemplate.setConfirmCallback(this::confirm);}/*交换机确认回调1.交换机收到了消息 触发回调1.1 correlationData(我们在发消息的时候自己创建的) 消息的ID以及消息内容1.2 ack 交换机收到消息 true1.3 cause 交换机收到消息的原因 null---------------------------------2.交换机未收到消息 触发回调2.1 correlationData 消息的ID以及消息内容2.2 ack 交换机未收到消息 false2.3 cause 失败原因*/Overridepublic 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.133port: 5672username: 2252631565password: 2252631565
# 高级发布确认 发布消息成功后将会触发回调方法publisher-confirm-type: correlated# 消息回退 当消息未路由至队列时触发publisher-returns: true
修改回调接口
实现 RabbitTemplate.ReturnsCallback 接口并实现方法
Slf4j
Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{Autowiredprivate RabbitTemplate rabbitTemplate;PostConstructpublic void init(){//注入rabbitTemplate.setConfirmCallback(this::confirm);rabbitTemplate.setReturnsCallback(this::returnedMessage);}/*交换机确认回调1.交换机收到了消息 触发回调1.1 correlationData(我们在发消息的时候自己创建的) 消息的ID以及消息内容1.2 ack 交换机收到消息 true1.3 cause 交换机收到消息的原因 null---------------------------------2.交换机未收到消息 触发回调2.1 correlationData 消息的ID以及消息内容2.2 ack 交换机未收到消息 false2.3 cause 失败原因*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){log.info(时间:{}发送消息到交换机成功,new Date());}else {log.info(发送消息到交换机失败原因为{},cause.toString());}}//当消息未路由到队列时触发 只有失败时才触发 若消息发送至延迟队列则一定会触发回退 记得根据交换机名称排除延迟队列Overridepublic 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 correlationData2new CorrelationData(2);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY123,message,correlationData2);}
测试 回退未进入队列的消息并重新发送消息。
二、备份交换机 什么是备份交换机呢备份交换机可以理解为 RabbitMQ 中交换机的“备胎”当我们为某一个交换机声明一个对应的备份交换机时就是为它创建一个备胎当交换机接收到一条不可路由消息时将会把这条消息转发到备份交换机中由备份交换机来进行转发和处理通常备份交换机的类型为 Fanout 扇出这样就能把所有消息都投递到与其绑定的队列中然后我们在备份交换机下绑定一个队列这样所有那些原交换机无法被路由的消息就会都进 入这个队列了。当然我们还可以建立一个报警队列用独立的消费者来进行监测和报警并可以重发消息。
实战
需要一个备份交换机 backup.exchange类型为 fanout该交换机发送消息到队列 backup.queue 和 warning.queue。 修改高级确认发布 配置类
Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGEconfirm.exchange;public static final String CONFIRM_QUEUEconfirm.queue;public static final String ROUTING_KEYkey1;//备份交换机public static final String BACKUP_EXCHANGEbackup.exchange;//备份队列public static final String BACKUP_QUEUEbackup.queue;//报警队列public static final String WARNING_QUEUEwarning.queue;Beanpublic DirectExchange confirmExchange(){//绑定确认交换机与备份交换机MapString,Object argumentnew HashMap();argument.put(alternate-exchange,BACKUP_EXCHANGE);return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).withArguments(argument).build();}//备份交换机Beanpublic FanoutExchange backupExchange(){return new FanoutExchange(BACKUP_EXCHANGE);}Beanpublic Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE).build();}//备份队列Beanpublic Queue backupQueue(){return QueueBuilder.durable(BACKUP_QUEUE).build();}//警告队列Beanpublic Queue warningQueue(){return QueueBuilder.durable(WARNING_QUEUE).build();}Beanpublic Binding EAndQBind(Qualifier(confirmExchange) DirectExchange exchange,Qualifier(confirmQueue)Queue queue){return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}//绑定备份交换机与两个队列Beanpublic Binding BAndBBing(Qualifier(backupExchange) FanoutExchange exchange,Qualifier(backupQueue)Queue queue){return BindingBuilder.bind(queue).to(exchange);}Beanpublic 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 correlationDatanew CorrelationData(1);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);//向一个不存在的队列发送消息CorrelationData correlationData2new CorrelationData(2);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY123,message,correlationData2);}
报警消费者
接收不可路由的消息
/*** 报警消费者*/
Slf4j
Component
public class WarningConsumer {//监听报警消息RabbitListener(queues ConfirmConfig.WARNING_QUEUE)public void receiveWarningMsg(Message message){String msgnew String(message.getBody());log.info(报警发现不可路由消息{},重发消息,msg);}
}测试 生产者发送两条消息 一个配置正确的路由另一个是错误的路由。预期目标是正确路由正常接收消息错误路由传输的信息由警告队列接收。 在此案例中也设置了消息回退的配置但是没有触发消息回退。由此得出备份交换机的优先级更高。
