网站seo方案撰写,淄博建企业网站,天津市建设工程信息网官网首页,公司网站制作高端问题#xff1a;消息在转换机无法被路由
发布确认高级作用在生产者发送到转换机#xff0c;回退消息作用在消息在转换机无法被路由的情况#xff08;消息无法路由的意思是#xff0c;消息在转换机没有匹配到对应的队列#xff09;#xff0c;进行消息回退#xff0c;打…问题消息在转换机无法被路由
发布确认高级作用在生产者发送到转换机回退消息作用在消息在转换机无法被路由的情况消息无法路由的意思是消息在转换机没有匹配到对应的队列进行消息回退打印日志但增加了生产者的复杂性。 前面在设置死信队列的文章中我们提到可以为队列设置死信交换机来存储那些 处理失败的消息可是这些不可路由消息根本没有机会进入到队列因此无法使用死信队列来保存消息。
解决方法设置备份交换机
在 RabbitMQ 中有一种备份交换机的机制存在可以很好的应对这个问题。什么是备份交换机呢备份交换机可以理解为 RabbitMQ 中交换机的“备胎”当我们为某一个交换机声明一个对应的备份交换机时 就是为它创建一个备胎当交换机接收到一条不可路由消息时将会把这条消息转发到备份交换机中由 备份交换机来进行转发和处理通常备份交换机的类型为 Fanout 这样就能把所有消息都投递到与其绑 定的队列中然后我们在备份交换机下绑定一个队列这样所有那些原交换机无法被路由的消息就会都进入这个队列了。当然我们还可以建立一个报警队列用独立的消费者来进行监测和报警。
注若 转换机 设置 备份转换机在消息在转换机无法被路由的情况下就不会调用 回退消息 的方法消息 直接转发到备份转换机
可参考一下代码
package com.ai.boy.config;import com.ai.boy.common.RabbitMqUtils;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** RabbitMq配置类声明队列、交换机* 绑定备份转换机Fanout并绑定备份队列、警告队列* 备份转换机 绑定在 正常转换机 上*/
Configuration
public class RabbitMqConfig {/*** 声明交换机 Exchange* 并设置该交换机的备份交换机* */Bean(confirmExchange)public DirectExchange confirmExchange(){/*** return new DirectExchange(CONFIRM_EXCHANGE_NAME);* 若不设置备份交换机按照以上即可* */return ExchangeBuilder.directExchange(RabbitMqUtils.CONFIRM_EXCHANGE_NAME).durable(true)/**设置该交换机的备份交换机*/.withArgument(alternate-exchange, RabbitMqUtils.BACKUP_EXCHANGE_NAME).build();}/*** 声明确认队列* */Bean(confirmQueue)public Queue confirmQueue(){return QueueBuilder.durable(RabbitMqUtils.CONFIRM_QUEUE_NAME).build();}/*** 声明确认队列绑定关系* */Beanpublic Binding queueBinding(Qualifier(confirmQueue) Queue queue,Qualifier(confirmExchange) DirectExchange exchange){return BindingBuilder.bind(queue).to(exchange).with(RabbitMqUtils.CONFIRM_KEY);}/*** 声明备份交换机 Exchange* */Bean(backupExchange)public FanoutExchange backupExchange(){return new FanoutExchange(RabbitMqUtils.BACKUP_EXCHANGE_NAME);}/*** 声明备份队列* */Bean(backQueue)public Queue backQueue(){return QueueBuilder.durable(RabbitMqUtils.BACKUP_QUEUE_NAME).build();}/*** 声明警告队列* */Bean(warningQueue)public Queue warningQueue(){return QueueBuilder.durable(RabbitMqUtils.WARNING_QUEUE_NAME).build();}/*** 声明 备份队列 绑定关系* 备份队列绑定备份备份交换机* */Beanpublic Binding backupBinding(Qualifier(backQueue) Queue queue,Qualifier(backupExchange) FanoutExchange backupExchange){return BindingBuilder.bind(queue).to(backupExchange);}/*** 声明 报警队列 绑定关系* 报警队列绑定备份备份交换机* */Beanpublic Binding warningBinding(Qualifier(warningQueue) Queue queue,Qualifier(backupExchange) FanoutExchange backupExchange){return BindingBuilder.bind(queue).to(backupExchange);}}
package com.ai.boy.common;public class RabbitMqUtils {public static final String CONFIRM_EXCHANGE_NAME confirm.exchange;public static final String CONFIRM_QUEUE_NAME confirm.queue;public static final String CONFIRM_KEY key1;public static final String BACKUP_EXCHANGE_NAME backup.exchange;public static final String BACKUP_QUEUE_NAME backup.queue;public static final String WARNING_QUEUE_NAME warning.queue;}
发布确认高级、回退消息
package com.ai.boy.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
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;/*** 发布确认高级、回退消息* 作用在生产者发送消息到转换机过程*/
Component
Slf4j
public class RabbitMqCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {Autowiredprivate RabbitTemplate rabbitTemplate;PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(this);/*** true* 交换机无法将消息进行路由时会将该消息返回给生产者* false* 如果发现消息无法进行路由则直接丢弃*/rabbitTemplate.setMandatory(true);rabbitTemplate.setReturnsCallback(this);}/*** 发布确认高级* 交换机不管是否收到消息的一个回调方法* 参数* CorrelationData 消息相关数据* ack 交换机是否收到消息*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String idcorrelationData!null?correlationData.getId():;if(ack){log.info(交换机已经收到 id 为:{}的消息,id);}else{log.info(交换机还未收到 id 为:{}消息,由于原因:{},id,cause);}}/*** 回退消息* 作用在消息在 转换机 无法被路由的情况下 执行 的回调方法* 发布确认高级 已确定消息已发送成功但在转换机的消息没有匹配到对应的队列* 注若 转换机 设置 备份转换机就不会调用该方法消息直接转发到备份转换机* */Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error(回退消息:{}被交换机{}退回退回原因 :{}路由key:{},new String(returnedMessage.getMessage().getBody()),returnedMessage.getExchange(),returnedMessage.getReplyText(),returnedMessage.getRoutingKey());}}
