逻辑网络设计的目标是什么?,网站seo外包价格,优化网站排名技巧,WordPress自动readmore在初学rabbitMq的时候#xff0c;伙伴们肯定已经接触到了“发布确认”的概念#xff0c;但是到了后期学习中#xff0c;会接触到“springboot”中使用“发布确认”高级的概念。后者主要是解决什么问题呢#xff1f;或者是什么样的场景引出这样的概念呢#xff1f; 在生产环…在初学rabbitMq的时候伙伴们肯定已经接触到了“发布确认”的概念但是到了后期学习中会接触到“springboot”中使用“发布确认”高级的概念。后者主要是解决什么问题呢或者是什么样的场景引出这样的概念呢 在生产环境中由于一些不明原因导致 rabbitmq 重启在 rabbitmq 重启期间生产者投递失败导致消息丢失需要手动处理和恢复。因此为了确保rabbitmq 的消息可靠投递,特别是在这样比较极端的情况rabbitmq 集群不可用的时候对无法投递的消息进行处理。废话不说直接开始撸代码在代码中解决实际问题~
一、代码架构分析 接触到这里对于一条完整的“rabbitmq消息”发布链的构成大家已经不陌生了。主要是由“消息生产者”、“交换机”、“队列”、“消费者”四个方面构成如图所示 二、构造“配置类”代码
声明交换机“confirm_exchange”、声明队列“confirm_queue”、通过routing-key对交换机和队列进行绑定。
package com.example.rabbitmq_demo.fabuquerengaoji;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** ClaseName: ConfirmConfig$* Description:配置类 发布确认高级* Author: wuhs* Date: 2023/8/16$ 14:32$* 快捷键ctrlshiftu 字母大小写转化*/
Configuration
public class ConfirmConfig {// 交换机public static final String CONFIRM_EXCHANGE_NAME confirm_exchange;// 队列public static final String CONFIRM_QUEUE_NAME confirm_queue;// ROUTING-KEYpublic static final String CONFIRM_ROUTING_KEY key1;Beanpublic DirectExchange confirmExchange() {return new DirectExchange(CONFIRM_EXCHANGE_NAME);}Beanpublic Queue confirmQueue() {return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();}Beanpublic Binding queueBindingExchange(Qualifier(confirmExchange) DirectExchange directExchange,Qualifier(confirmQueue) Queue queue) {//一般使用在项目中使用Qualifier来限定注入的Bean。return BindingBuilder.bind(queue).to(directExchange).with(CONFIRM_ROUTING_KEY);}
}三、构建消费者代码:
通过RabbitListener(queues ConfirmConfig.CONFIRM_QUEUE_NAME)来监听队列以此“充当”消费者。这一块也没啥好说的直接上代码
package com.example.rabbitmq_demo.fabuquerengaoji;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** ClaseName: Consumer$* Description:消费者* Author: wuhs* Date: 2023/8/16$ 15:18$】*/
Slf4j
Component
public class Consumer {RabbitListener(queues ConfirmConfig.CONFIRM_QUEUE_NAME)public void reciverConfirmMessage(Message message) {String msg new String(message.getBody());log.info(接收到的队列confirm.queue消息{}, msg);}
}四、创建“回调”方法 在最开始我们说到“确保rabbitmq 的消息可靠投递”的概念那么具体如何确保呢如果我们在消费者每次消费成功、未消费成功交换机都能进行“回调”确认是不是就能知道哪些消息消费成功、哪些没有消费成功呢 在RabbitTemplate中有一个方法接口ConfirmCallback我们只需要实现这个接口并实现“confirm”方法并将它注入进RabbitTemplate工具中即可创建“回调”。具体代码如下
package com.example.rabbitmq_demo.fabuquerengaoji;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;/*** ClaseName: MyCallBack$* Description:* Author: wuhs* Date: 2023/8/16$ 16:17$*/
Slf4j
Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {Autowiredprivate RabbitTemplate rabbitTemplate;PostConstructpublic void init() {// 注入rabbitTemplate.setConfirmCallback(this);}//交换机确认回调方法 Overridepublic void confirm(CorrelationData correlationData, boolean ack, String reason) {String id correlationData ! null ? correlationData.getId() : ;if (ack) {//发消息 交换机接收到了消息 回调log.info(交换机已经收到了ID为{}的消息, id);} else {//发消息 交换机没有接收到了消息 回调log.info(交换机没有收到了ID为{}发的的消息失败的原因是{}, id, reason);}}
}confirm方法参数介绍 * 1. correlationData 保存回调消息的ID及相关信息
* 2. 交换机是否收到了消息 acktrue收到、ackfalse未收到
* 3. reason 失败的原因六、配置类声明application.yml 在这里需要注意这也是最容易踩得坑不知道有没有小伙伴遇没遇到“publisher-confirm-type: correlated”也声明了但是项目创建启动发布消息之后“没有成功回调”的情况查看了很多的文章很多博主只配置了publisher-confirm-type、但是并没有开启“confirm 确认机制”所以会存在“误导”导致一直找不到失败的原因~具体正确配置看代码:
server:port: 8899spring:rabbitmq:host: 124.221.94.214port: 5672username: xgsmpassword: xgsm123# 发送者开启 confirm 确认机制publisher-confirms: truepublisher-confirm-type: correlatedpublisher-confirm-type参数介绍 publisher-confirm-type这个参数一共有三种配置方法 # NONE:禁用发布确认是默认值。 # CORRELATED:发布消息后交换机会触发回调方法。 # SIMPLE:有两种效果 1和CORRELATED一样会触发回调方法 2发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或waitForConfirmsOrDie 方法等待 broker 节点返回发送结果根据返回结果来判定下一步的逻辑 # 要注意的点是waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel则接下来无法发送消息到 broker。 七、创建Controller层消息生产者 这里演示三种情况。第一种为正常情况下发送成功后的回调第二种消息为发送失败、当交换机不存在则发送失败模拟发送失败所以将交换机名称修改即可
package com.example.rabbitmq_demo.fabuquerengaoji;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.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;/*** ClaseName: ProducerController$* Description:消息生产者* Author: wuhs* Date: 2023/8/16$ 14:58$*/
Slf4j
RestController
RequestMapping(/confirm)
public class ProducerController {Autowiredprivate RabbitTemplate rabbitTemplate;// PathVariable主要作用映射URL绑定的占位符RequestMapping(/sendMessage/{message})public void sendMessage(PathVariable String message) {//正常发送CorrelationData correlationData new CorrelationData(1);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.CONFIRM_ROUTING_KEY, message,correlationData);//发送失败-交换机不存在的情况CorrelationData correlationData2 new CorrelationData(2);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME2, ConfirmConfig.CONFIRM_ROUTING_KEY, message,correlationData2);log.info(发送的消息为{}, message);}
}测试结果 如果是routing-key错误这种情况会触发回调嘛让我们验证一下修改routing-key为“错误值” CorrelationData correlationData3 new CorrelationData(3);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.CONFIRM_ROUTING_KEYkey3, message,correlationData3);
测试结果 通过结果可以看出消息发送成功了而且也触发了“成功的回调”。但是我们知道的是由于路由失败这里消费者并没有对消息进行消费这是为什么呢那是因为在仅开启了生产者确认机制的情况下交换机接收到消息后会直接给消息生产者发送确认消息如果发现该消息不可路由那么消息会被直接丢弃此时生产者是不知道消息被丢弃这个事件的。解决方式为通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。具体操作如下
1、application.yml文件中添加消息回退配置
# 发送者开启 return 确认机制publisher-returns: true 2、实现RabbitTemplate中的方法接口ReturnCallback并实现“returnedMessage”方法最后将类注入到RabbitTemplate的RabbitTemplate中详细代码如下
package com.example.rabbitmq_demo.fabuquerengaoji;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
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;/*** ClaseName: MyCallBack$* Description:* Author: wuhs* Date: 2023/8/16$ 16:17$*/
Slf4j
Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {Autowiredprivate RabbitTemplate rabbitTemplate;PostConstructpublic void init() {// 注入rabbitTemplate.setConfirmCallback(this);rabbitTemplate.setReturnCallback(this);}/*** 交换机确认回调方法* 1、发消息 交换机接收到了消息 回调* 1.1 correlationData 保存回调消息的ID及相关信息* 1.2 交换机收到消息 acktrue* 2、发消息 交换机接收失败了 回调* 2.1 correlationData 保存回调消息的ID及相关信息* 2.2 交换机接收到消息 ackfalse* 2.3 reason 失败的原因*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String reason) {String id correlationData ! null ? correlationData.getId() : ;if (ack) {log.info(交换机已经收到了ID为{}的消息, id);} else {log.info(交换机没有收到了ID为{}发的的消息失败的原因是{}, id, reason);}}//可以在当消息传递的过长中不可达目的地时将消息返回给生产者// 只有不可待目的地的时候 才进行回退Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info(消息{},被交换机{}退回,退回原因:{},路由key:{}, message, exchange, replyText, routingKey);}
}测试结果 2023-08-17 10:36:32.476 INFO 21108 --- [221.94.214:5672] c.e.r.fabuquerengaoji.MyCallBack : 消息(Body:消息确认发布测试 MessageProperties [headers{spring_returned_message_correlation3}, contentTypetext/plain, contentEncodingUTF-8, contentLength0, receivedDeliveryModePERSISTENT, priority0, deliveryTag0]),被交换机confirm_exchange退回,退回原因:NO_ROUTE,路由key:key1key3 问题解决~
