余姚建设网站,网站是怎么做新手引导,抖音代运营一般哪家好,前端网页模板说明#xff1a;在RabbitMQ消息传递过程中#xff0c;有以下问题#xff1a; 消息没发到交换机 消息没发到队列 MQ宕机#xff0c;消息在队列中丢失 消息者接收到消息后#xff0c;未能正常消费#xff08;程序报错#xff09;#xff0c;此时消息已在队列中移除
…说明在RabbitMQ消息传递过程中有以下问题 消息没发到交换机 消息没发到队列 MQ宕机消息在队列中丢失 消息者接收到消息后未能正常消费程序报错此时消息已在队列中移除
针对以上问题提供以下解决方案 消息确认确认消息是否发送到交换机、队列 消息持久化持久化消息以防MQ宕机造成消息丢失 消费者消息确认确认消费者已正确消费消息才把消息从队列中删除 消息确认
可以使用Rabbit MQ提供的publisher confirm机制来避免消息发送到MQ过程丢失。具体实现是publisher-confirm发送者确定、publisher-return发送者回执前者判断消息到交换机、后者判断交换机到队列 publisher-confirm发送者确定 消息成功投递到交换机返回ack 消息未投递到交换机返回nack
publisher-return发送者回执
消息投递到交换机但没有到队列返回ack即失败原因 在生产者端添加配置
spring:rabbitmq:# rabbitMQ相关配置host: 118.178.228.175port: 5672username: rootpassword: 123456virtual-host: /# 开启生产者确认correlated为异步simple为同步publisher-confirm-type: correlated# 开启publish-return功能基于callback机制publisher-returns: true# 开启消息路由失败的策略true是调用returnCallback方法false是丢弃消息template:mandatory: truepublisher-return发送者回执代码
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;/*** 发送者回执实现*/
Slf4j
Configuration
public class CommonConfig implements ApplicationContextAware {Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplate对象RabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*** 回执信息* param message 信息对象* param replyCode 回执码* param replyText 回执内容* param exchange 交换机* param routingKey 路由键值*/Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info(消息发送队列失败replyCode{}replyText{}exchange{}routingKey{}message{},replyCode,replyText,exchange,routingKey,message);}});}
}publisher-confirm发送者确定代码 Testpublic void sendExceptionMessage() {// 路由键值String routingKey exception;// 消息String message This is a exception message;// 给消息设置一个唯一IDCorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());// 编写confirmCallBack回调函数correlationData.getFuture().addCallback(new SuccessCallbackCorrelationData.Confirm() {Overridepublic void onSuccess(CorrelationData.Confirm confirm) {if (confirm.isAck()) {// 消息发送交换机成功log.debug(消息送达至交换机成功);} else {// 消息发送交换机失败打印消息log.error(消息未能送达至交换机ID{}原因{}, correlationData.getId(), confirm.getReason());}}}, new FailureCallback() {// 消息发送交换机异常Overridepublic void onFailure(Throwable ex) {log.error(消息发送交换机异常ID:{}原因{}, correlationData.getId(), ex.getMessage());}});rabbitTemplate.convertAndSend(amq.direct, routingKey, message, correlationData);}测试设置一个不存在的routingKey被发送者确认publisher-confirm捕获到
// 路由键值
String routingKey null;设置一个不存在的路由被发送者回执publisher-return捕获到
rabbitTemplate.convertAndSend(null, routingKey, message, correlationData);消息持久化
消息持久化是指把消息保存到磁盘中在RabbitMQ宕机或者关机时重启后消息仍可以保存下来。消息依赖于交换机、队列因此持久化消息同时也需要持久化交换机、队列。
创建一个持久化的交换机、队列
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 消息持久化*/
Configuration
public class DurableConfig {/*** 交换机持久化* return*/Beanpublic DirectExchange directExchange(){// 三个参数分别是交换机名、是否持久化、没有队列与之绑定时是否自动删除return new DirectExchange(durable.direct,true,false);}/*** 队列持久化* return*/Beanpublic Queue durableQueue(){return QueueBuilder.durable(durable.queue).build();}/*** 交换机与队列绑定* return*/Beanpublic Binding binding(){return BindingBuilder.bind(durableQueue()).to(directExchange()).with(durable);}}发送一个持久化的消息 /*** 发送持久化消息*/Testpublic void sendDurableMessage() {String routingKey durable;CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());Message message MessageBuilder.withBody(This is a durable message.getBytes(StandardCharsets.UTF_8))// 设置该消息未持久化消息.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();rabbitTemplate.convertAndSend(durable.direct, routingKey, message, correlationData);}打开RabbitMQ管理平台可以看到delivery_mode: 2表示该消息是持久化消息 源码MessageDeliveryMode类
实际上交换机、队列默认就是持久化的durable: true所以不用特意设置 消费者消息确认
介绍
消费者消息确认是为了确保消费者已经消费了消息才让MQ把该消息删除
可通过在消费者的配置文件中增加下面这行配置实现备选项有以下三个 none关闭ack表示不做处理消息发给消费者之后就立即被删除 auto自动ack表示由Spring检测代码是否出现异常出现异常则保留消息没有异常则删除消息 manual手动ack可根据业务手动编写代码返回ack
spring:rabbitmq:listener:simple:# 设置消息确认模式acknowledge-mode: none测试none
可编写代码测试下面是生产者代码发送消息 /*** 发送普通消息*/Testpublic void sendNoneMessage() {String directName none.direct;String routingKey none;String message This is a test message;rabbitTemplate.convertAndSend(directName, routingKey, message);}消费者代码有问题未能正常消费消息 RabbitListener(bindings QueueBinding(value Queue(name none.queue),exchange Exchange(name none.direct,type ExchangeTypes.DIRECT),key {none}))public void getNoneMessage(String normalMessage){System.out.println(1/0);System.out.println(normalMessage normalMessage);}测试结果程序报错消息也没能保留下来 测试auto
更改设置为auto重试 但是消息未被删除 这种情况在实际开发中是不能允许可以通过更改消费失败的重试机制解决。
消费失败重试机制
方法一设置retry
因为消息被消费失败消息会一直循环重试无限循环导致mq的消息处理飙升带来不必要的压力这种情况可以通过在消费者端添加以下配置限制失败重试的条件来解决
spring:rabbitmq:listener:simple:retry:# 开启消费者失败重试enabled: true# 初次失败等待时长为1秒initial-interval: 1000# 失败的等待时长倍数即后一次等待的时间是前一次等待时间的多少倍multiplier: 1# 最多重试次数max-attempts: 3# true 无状态 false 有状态 如果业务中包含事务 改为falsestateless: true开启后控制台可以发现信息不回一直循环打印而是打印数条后停止日志信息中有提示“Retry Policy Exhausted”重试策略已用尽 这种通过配置的方式并不会重试数次后仍保留消息而是重试数次仍失败随即丢弃消息消息丢失这在实际开发中也是不能被允许的。
方法二路由存储消息
因此可以通过下面这个方法把消费失败的消息通过交换机路由到另外的队列中存储起来等业务代码被修复再路由回来消费。 代码如下
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 错误消息队列*/
Configuration
public class ErrorMessageQueueConfig {/*** 创建一个交换机用于路由消费失败的消息* return*/Beanpublic DirectExchange errorExchange(){return new DirectExchange(error.direct);}/*** 创建一个队列用于存储消费失败的消息* return*/Beanpublic Queue errorQueue(){return new Queue(error.queue);}/*** 绑定* return*/Beanpublic Binding errorBinding(){return BindingBuilder.bind(errorQueue()).to(errorExchange()).with(error);}/*** 路由当消费失败时把消费失败的消息路由到此队列中路由key为error* param rabbitTemplate* return*/Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate,error.direct,error);}
}可以看到消息消费失败后并没有被丢失而是路由到错误队列中存储了起来。因为错误队列没有设置RabbitListener所以可以存储消息等带代码问题被排查出来后可以再针对该队列设置监听方法消费这部分错误的消息。 另外值得一提的是消费者这边的控制台会报一个警告提示路由密钥错误。我们可以理解在RabbitMQ底层会把消费失败了的消息统一路由到一个地方去而我们这种手动把消费失败的消息路由到自定义的队列中的方式打破了这种“默认的规则”所以报了一个这样的警告。这种警告是在可控范围内的。 总结
RabbitMQ发送消息为了确保消息的可靠性保证消息能被交换机、队列收到消息能被正常消费而不会因消费失败而丢失提供了对应的一系列方法并且最后还提供了两种消费失败重试方法优化了消费过程非常Nice。
