最流行网站开发工具,哪个网站建设,做的网站 只显示代码,好听的公司名字大全集1.前言
在生产环境中由于一些不明原因#xff0c;导致 RabbitMQ 重启#xff0c;在 RabbitMQ 重启期间生产者消息投递失败#xff0c; 导致消息丢失#xff0c;需要手动处理和恢复。于是#xff0c;我们开始思考#xff0c;如何才能进行 RabbitMQ 的消息可靠投递呢…1.前言
在生产环境中由于一些不明原因导致 RabbitMQ 重启在 RabbitMQ 重启期间生产者消息投递失败 导致消息丢失需要手动处理和恢复。于是我们开始思考如何才能进行 RabbitMQ 的消息可靠投递呢
2.添加配置信息
在application.properties文件中添加如下配置交换机开启消息确认模式
#NONE 值是禁用发布确认模式是默认值
#CORRELATED 值是发布消息成功到交换器后会触发回调方法
#SIMPLE 值经测试有两种效果其一效果和 CORRELATED 值一样会触发回调方法
# 其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果
# 根据返回结果来判定下一步的逻辑要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel
# 则接下来无法发送消息到 broker;
spring.rabbitmq.publisher-confirm-typecorrelatedNONE 值是禁用发布确认模式是默认值CORRELATED 值是发布消息成功到交换器后会触发回调方法SIMPLE 值经测试有两种效果其一效果和 CORRELATED 值一样会触发回调方法其二在发布消息成功后使用rabbitTemplate 调用waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker节点返回发送结果根据返回结果来判定下一步的逻辑要注意的点是waitForConfirmsOrDie 方法如果返回 false则会关闭 channel则接下来无法发送消息到 broker;
3. 配置类
package com.hong.springboot.rabbitmq.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** Description: 发布确认高级版配置类* Author: hong* Date: 2024-03-05 20:52* Version: 1.0**/
Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE_NAME confirm.exchange;public static final String CONFIRM_QUEUE_NAME confirm.queue;public static final String CONFIRM_ROUTING_KEY key1;//声明业务 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(CONFIRM_ROUTING_KEY);}
}4.生产者
package com.hong.springboot.rabbitmq.controller;import com.hong.springboot.rabbitmq.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.text.SimpleDateFormat;
import java.util.Date;/*** Description: 发布确认高级版生产者* Author: hong* Date: 2024-03-05 20:58* Version: 1.0**/
Slf4j
RequestMapping(/confirm/)
RestController
public class ConfirmProducerController {Autowiredprivate RabbitTemplate rabbitTemplate;//http://localhost:8080/confirm/sendMsg/Hi,JAVA小生不才GetMapping(sendMsg/{message})public void sendMsg(PathVariable String message) {log.info(当前时间{},发送信息给队列:{},new SimpleDateFormat(yyyy-MM-dd HH:mm:ss).format(new Date()) , message);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.CONFIRM_ROUTING_KEY, message);}
}5.消费者
package com.hong.springboot.rabbitmq.consumer;import com.hong.springboot.rabbitmq.config.ConfirmConfig;
import com.hong.springboot.rabbitmq.config.DelayedQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.text.SimpleDateFormat;
import java.util.Date;/*** Description: 发布确认高级版消费者* Author: hong* Date: 2024-03-05 21:05* Version: 1.0**/
Slf4j
Component
public class ConfirmConsumer {RabbitListener(queues ConfirmConfig.CONFIRM_QUEUE_NAME)public void receiveConfirmMessage(Message message){String msg new String(message.getBody());log.info(当前时间{},收到信息{},new SimpleDateFormat(yyyy-MM-dd HH:mm:ss).format(new Date()) , msg);}
}正常情况下发送http://localhost:8080/confirm/sendMsg/Hi,JAVA小生不才
6.回调接口
package com.hong.springboot.rabbitmq.config;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;/*** Description: 发布确认高级版消息生产者的回调接口* Author: hong* Date: 2024-03-09 21:58* Version: 1.0**/
Slf4j
Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback{Autowiredprivate RabbitTemplate rabbitTemplate;PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(this);}/*** 交换机不管是否收到消息的一个回调方法* 1.收到消息* correlationData 保存回调消息的id及相关信息* b true 交换机收到消息* s null* 2.未收到消息* correlationData 保存回调消息的id及相关信息* b false 交换机未收到消息* s 失败的原因* param correlationData 消息相关数据* param b 交换机是否收到消息* param s 没收到消息的原因*/Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {String id correlationData ! null ? correlationData.getId() : ;if (b) {log.info(交换机已经收到id为:{}的消息, id);} else {log.info(交换机还未收到id为:{}消息原因:{}, id, s);}}
}修改ConfirmProducerController中sendMsg方法 交换机改个名字模拟交换机收不到消息 GetMapping(sendMsg/{message})public void sendMsg(PathVariable String message) {CorrelationData correlationData new CorrelationData(1);log.info(当前时间{},发送信息给队列:{},new SimpleDateFormat(yyyy-MM-dd HH:mm:ss).format(new Date()) , message);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME123, ConfirmConfig.CONFIRM_ROUTING_KEY, message,correlationData);}将routingKey改个名字模拟队列收不到消息 GetMapping(sendMsg/{message})public void sendMsg(PathVariable String message) {CorrelationData correlationData1 new CorrelationData(1);log.info(当前时间{},发送信息给队列:{},new SimpleDateFormat(yyyy-MM-dd HH:mm:ss).format(new Date()) , message----ConfirmConfig.CONFIRM_ROUTING_KEY);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.CONFIRM_ROUTING_KEY,message----ConfirmConfig.CONFIRM_ROUTING_KEY,correlationData1);CorrelationData correlationData2 new CorrelationData(2);log.info(当前时间{},发送信息给队列:{},new SimpleDateFormat(yyyy-MM-dd HH:mm:ss).format(new Date()) , message----ConfirmConfig.CONFIRM_ROUTING_KEYabc);rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME, ConfirmConfig.CONFIRM_ROUTING_KEYabc,message----ConfirmConfig.CONFIRM_ROUTING_KEYabc,correlationData2);}7.回退消息
从以上模拟场景可以看出在仅开启生产者确认机制交换机接收到消息后会直接给生产者发送确认消息但若发现该消息不可路由那么消息会被直接丢弃此时生产者是不知道消息被丢弃的。因此我们借用mandatory参数在当消息传递过程中不可达目的地时将消息返回给生产者。
7.1.开启消息回退机制
配置文件中添加如下配置
#开启消息回退机制
spring.rabbitmq.publisher-returnstrue7.2. 添加消息回退回调 PostConstructpublic void init(){rabbitTemplate.setConfirmCallback(this);rabbitTemplate.setReturnsCallback(this);}/*** 当消息传递过程中不可达目的地时将消息返回给生产者* 只有不可达目的地时才回调* param returnedMessage*/Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error(消息{}被交换机 {} 退回原因{}路由key{},code:{},new String(returnedMessage.getMessage().getBody()), returnedMessage.getExchange(),returnedMessage.getReplyText(), returnedMessage.getRoutingKey(), returnedMessage.getReplyCode());}
