个人或主题网站建设,南京做南京美容整形网站,万网网站建设教程,自动生成网页代码的软件发布确认springboot版本
确认机制方案#xff1a; 代码架构图#xff1a; 配置文件#xff1a;
在application.properties全局配置文件中添加spring.rabbitmq.publish-confirm-type属性#xff0c;这个属性有以下几种值
none:禁用发布确认模式(默认)0 correlated:发布消…发布确认springboot版本
确认机制方案 代码架构图 配置文件
在application.properties全局配置文件中添加spring.rabbitmq.publish-confirm-type属性这个属性有以下几种值
none:禁用发布确认模式(默认)0 correlated:发布消息成功到交换机后会触发回调方法 simple有两种效果 第一种效果是和correlated一样会触发回调方法 第二种效果是在发布消息成功以后使用rabbitTemplate调用waitForConfirms或者waitForConfirmsOrDie方法等待broker节点返回发送结果根据返回结果来判单下一步的逻辑 waitForConfirmsOrDie方法如果返回false则会关闭信道那么接下来就无法发送消息到broker
spring.rabbitmq.host182.92.234.71
spring.rabbitmq.port5672
spring.rabbitmq.usernameadmin
spring.rabbitmq.password123
spring.rabbitmq.publisher-confirm-typecorrelated
添加配置类
Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE_NAME confirm.exchange;public static final String CONFIRM_QUEUE_NAME confirm.queue;//声明业务 ExchangeBean(confirmExchange)public DirectExchange confirmExchange(){return new DirectExchange(CONFIRM_EXCHANGE_NAME);}// 声明确认队列Bean(confirmQueue)public Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();}// 声明确认队列绑定关系Beanpublic Binding queueBinding(Qualifier(confirmQueue) Queue queue,Qualifier(confirmExchange) DirectExchange exchange){return BindingBuilder.bind(queue).to(exchange).with(key1);}
}
消息生产者
RestController
RequestMapping(/confirm)
Slf4j
public class Producer {public static final String CONFIRM_EXCHANGE_NAME confirm.exchange;Autowiredprivate RabbitTemplate rabbitTemplate;Autowiredprivate MyCallBack myCallBack;//依赖注入 rabbitTemplate 之后再设置它的回调对象PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(myCallBack);}GetMapping(sendMessage/{message})public void sendMessage(PathVariable String message){//指定消息 id 为 1CorrelationData correlationData1new CorrelationData(1);String routingKeykey1;rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey,messageroutingKey,correl
ationData1);CorrelationData correlationData2new CorrelationData(2);routingKeykey2;rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME,routingKey,messageroutingKey,correl
ationData2);log.info(发送消息内容:{},message);}
}
回调接口
Component
Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {/*** 交换机不管是否收到消息的一个回调方法* 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);}}
}
消息消费者
Component
Slf4j
public class ConfirmConsumer {public static final String CONFIRM_QUEUE_NAME confirm.queue;RabbitListener(queues CONFIRM_QUEUE_NAME)public void receiveMsg(Message message){String msgnew String(message.getBody());log.info(接受到队列 confirm.queue 消息:{},msg);}
}
结果分析 可以看到发送了两条消息第一条消息的 RoutingKey 为 key1 第二条消息的 RoutingKey 为 key2两条消息都成功被交换机接收也收到了交换机的确认回调但消费者只收到了一条消息因为 第二条消息的 RoutingKey 与队列的 BindingKey 不一致也没有其它队列能接收这个消息所有第二条 消息被直接丢弃了。 回退消息
mandatory参数
如果我们仅仅开启了生产者确认机制那么当交换机接受到消息后会直接给生产者发送确认消息但是如果发现消息不可以路由就会直接把消息丢弃此时消费者接受不到消息而且这个时候生产者也不知道消息被丢弃了这样就导致消息丢失我们可以通过设置mandatory参数是的消息在传递过程中出现不可到达的目的的时候可以把消息返回给生产者
Slf4j
Component
public class MessageProducer implements RabbitTemplate.ConfirmCallback ,
RabbitTemplate.ReturnCallback {Autowiredprivate RabbitTemplate rabbitTemplate;//rabbitTemplate 注入之后就设置该值PostConstructprivate void init() {rabbitTemplate.setConfirmCallback(this);/*** true* 交换机无法将消息进行路由时会将该消息返回给生产者* false* 如果发现消息无法进行路由则直接丢弃*/rabbitTemplate.setMandatory(true);//设置回退消息交给谁处理rabbitTemplate.setReturnCallback(this);}GetMapping(sendMessage)
public void sendMessage(String message){//让消息绑定一个 id 值CorrelationData correlationData1 new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(confirm.exchange,key1,messagekey1,correlationData1)
;log.info(发送消息 id 为:{}内容为{},correlationData1.getId(),messagekey1);CorrelationData correlationData2 new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(confirm.exchange,key2,messagekey2,correlationData2)
;log.info(发送消息 id 为:{}内容为{},correlationData2.getId(),messagekey2);
}Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String id correlationData ! null ? correlationData.getId() : ;if (ack) {log.info(交换机收到消息确认成功, id:{}, id);} else {log.error(消息 id:{}未成功投递到交换机,原因是:{}, id, cause);}}Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String
exchange, String routingKey) {log.info(消息:{}被服务器退回退回原因:{}, 交换机是:{}, 路由 key:{},new String(message.getBody()),replyText, exchange, routingKey);}
}
回调接口
Component
Slf4j
public class MyCallBack implements
RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {/*** 交换机不管是否收到消息的一个回调方法* 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(Message message, int replyCode, String replyText, String
exchange, String routingKey) {log.error( 消 息 {}, 被交换机 {} 退回退回原因 :{}, 路 由 key:{},new
String(message.getBody()),exchange,replyText,routingKey);}
}
结果分析
