赞
踩
以支付服务为案例进行分析
问题一、在扣减余额和更新支付状态之后,支付服务会发送消息通知交易服务更新状态,如果在支付服务与MQ之间出现了网络故障,消息无法到达MQ,交易服务无法收到消息。即发送者弄丢了消息。
问题二、消息投递到了MQ,但在MQ将消息传递给交易服务之前,MQ挂掉了,消息也会丢失。即MQ弄丢了消息。
问题三、消息投递到了交易服务,但在处理的过程中抛出异常,也会丢失消息。即消费者弄丢了消息。
消息的可靠性:消息被发送后,至少要被消费一次。
发送者的可靠性问题通过两种方式解决:生产者重连和生产者确认。
有的时候由于网络波动,可能会出现客户端连接MQ失败的情况。通过配置我们可以开启连接失败后的重连机制。
注意:当网络不稳定的时候,利用重试机制可以有效提高消息发送的成功率。不过Spring AMQP提供的重试机制是阻塞式的重试,也就是说多次重试等待的过程中,当前线程是被阻塞的,后续业务代码不会执行,会影响业务性能。如果对于业务性能有要求,建议禁用重试机制。如果一定要使用,请合理配置等待时长和重试次数,当然也可以考虑使用异步线程来执行发送消息的代码。
- spring:
- rabbitmq:
- connection-timeout: 1s #设置MQ的连接超时时间
- template:
- retry:
- enabled: true #开启超时重试机制
- initial-interval: 1000ms #失败后的初始等待时间
- multiplier: 1 # 失败后下次的等待时长倍数,下次等待时长 = initial-interval * multiplier
- max-attempts: 3 #最大重试次数
RabbitMQ提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后,在MQ成功收到消息后会返回确认消息给生产者。返回的结果有以下几种情况:
1、消息投递到了交换机,但是路由失败(routingKey不对,找不到对应的MQ)。此时会通过Publisher Return返回路由异常原因,然后返回ACK,告知投递成功。
2、临时消息投递到了MQ,并且入队成功,返回ACK,告知投递成功。
3、持久消息投递到了MQ,并且入队完成持久化(保存到磁盘),返回ACK ,告知投递成功。
4、其它情况都会返回NACK,告知投递失败。
生产者确认需要额外的网络和系统资源开销,尽量不要使用。
如果一定要使用,无需开启Publisher-Return机制,因为一般路由失败是自己业务问题。
对于nack消息可以有限次数重试,依然失败则记录异常消息。
第一步、添加yml配置
- spring:
- rabbitmq:
- #开启publisher confirm机制,并设置confirm类型:
- #1、none:关闭confirm机制;2、simple:同步阻塞等待MQ的回执消息;3、correlated:MQ异步回调方式返回回执消息
- publisher-confirm-type: correlated
- publisher-returns: true # 开后publisher return机制
第二步、配置Publisher-Return机制,每个RabbitTemplate只能配置一个ReturnCallback,因此需要在项目启动过程中配置
- @Slf4j
- @Configuration
- public class MqConfirmConfig implements ApplicationContextAware {
- @Override
- public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
- RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
- //配置回调
- rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
- @Override
- public void returnedMessage(ReturnedMessage returnedMessage) {
- log.debug("收到消息的return callback, exchange:{},key:{},msg:{},code:{},text:{}",
- returnedMessage.getExchange(),returnedMessage.getRoutingKey(),returnedMessage.getMessage(),
- returnedMessage.getReplyCode(),returnedMessage.getReplyText());
-
- }
- });
- }
- }
第三步、Publisher Confirm机制,在业务代码中主动获取异步结果
- @Test
- void testConfirmCallback() throws InterruptedException {
- //1、创建cd
- CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
- //2、添加ConfirmCallback
- cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
- @Override
- public void onFailure(Throwable ex) {
- log.error("消息回调失败",ex);
- }
-
- @Override
- public void onSuccess(CorrelationData.Confirm result) {
- log.debug("收到confirm callback 回执");
- if (result.isAck()){
- //消息发送成功
- log.debug("消息发送成功,收到ack");
- }else {
- //消息发送失败
- log.error("消息发送失败,收到nack,原因:" ,result.getReason());
-
-
- //重发消息逻辑代码
- }
- }
- });
- rabbitTemplate.convertAndSend("hmall.direct","red","hello",cd);
- Thread.sleep(2000);
- }
在默认情况下,RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题:
1、一旦MQ宕机,内存中的消息会丢失。
2、内存空间有限,当消费者故障或处理过慢时,会导致消息积压,引发MQ阻塞。
解决方法:数据持久化(3.6版本之前);Lazy Queue(3.6版本之后)。
RabbitMQ实现数据持久化包括3个方面:
交换机持久化(Durable属性,Spring默认设置为Durable);
队列持久化(Durable属性,Spring默认设置为Durable);
消息持久化(发送消息时设置delivery_mode=2persisent,Spring发送的消息默认是持久化的);
- @Test
- void testPageOut(){
- Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))
- .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();
- for (int i = 0; i < 1000000; i++) {
- rabbitTemplate.convertAndSend("simple.queue",message);
- }
- }
发送消息的情况
非持久化,会出现paged out,会阻塞IO,性能下降
持久化消息,不会出现阻塞情况,性能下降的比较小
从RabbitMQ的3.6.0版本开始,就增加了Lazy Queue的概念,也就是惰性队列。
惰性队列的特征如下:
1、接收到消息后直接存入磁盘而非内存(内存中只保留最近的消息,默认2048条)。2、消费者要消费消息时才会从磁盘中读取并加载到内存。
3、支持数百万条的消息存储。
4、在3.12版本后,所有队列都是Lazy Queue模式,无法更改 。
(3.12之前)需要设置一个队列为惰性队列,只需要在声明队列时,指定x-queue-mode属性为lazy即可:
- //声明Bean
- @Bean
- public Queue lazyQueue(){
- return QueueBuilder.durable("lazy.queue").lazy().build(); //创建惰性队列
- }
-
-
- //基于注解创建
- @RabbitListener(queuesToDeclare = @Queue(
- name = "lazy.queue",
- durable = "true",
- arguments = @Argument(name = "x-queue-mode",value = "lazy")
- ))
- public void listenLazyQueue(String msg){
- log.info("接收到lazy.queue的消息:{}",msg);
- }
发送非持久化消息,不会发生阻塞IO,消息直接到Paged Out,不用经过Memery到Paged Out的过程。
RabbitMQ如何保证消息的可靠性:
首先通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘,MQ重启消息依然存在。
RabbitMQ在3.6版本引入了LazyQueue,并且在3.12版本后会称为队列的默认模式,LazyQueue会将所有消息都持久化。
开启持久化和生产者确认时,RabbitMQ只有在消息持久化完成后才会给生产者返回ACK回执。
保证消费者的可靠性主要有三种手段:消费者确认机制;消费失败处理;业务幂等性
为了确认消费者是否成功处理消息,RabbitMQ提供了消费者确认机制(Consumer Acknowledgement)。当消费者处理消息结束后,应该向RabbitMQ发送一个回执,告知RabbitMQ自己消息处理状态。
回执有三种可选值:
ack:成功处理消息,RabbitMQ从队列中删除该消息。
nack:消息处理失败,RabbitMQ需要再次投递消息 。
reject:消息处理失败并拒绝该消息,RabbitMQ从队列中删除该消息(一般是消息的参数不正确) 。
Spring AMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式,有三种方式:
none:不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全,不建议使用
manual:手动模式。需要自己在业务代码中调用api,发送ack或reject,存在业务入侵,但更灵活
auto:自动模式。Spring AMQP利用AOP对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack。当业务出现异常时,根据异常判断返回不同结果:
如果是业务异常,会自动返回nack
如果是消息处理或校验异常,自动返回reject
第一步、开启消费者确认其机制
- spring:
- rabbitmq:
- listener:
- simple:
- prefetch: 1 #每次只能获取一条消息,处理完才能获取下一条消息
- acknowledge-mode: auto #none:关闭ack;manual:手动ack;auto:自动ack
第二步、消费者业务模拟异常
- @RabbitListener(queues = "simple.queue")
- public void listSimpleQueue(String msg) {
- System.out.println("消费者收到了simple.queue的消息:【" + msg + "】");
- throw new RuntimeException("测试异常");
- //throw new MessageConversionException("消息转换异常");
- }
第三步、发送消息
- @Test
- void testSendMessage2Queue() {
- //队列名称
- String queueName = "simple.queue";
- //消息
- String msg = "hello, amqp!";
- //发送消息
- rabbitTemplate.convertAndSend(queueName, msg);
- }
auto模式下,消息未被处理会保留消息,尝试重新投递给消费者。像模拟的异常情况下,消息会一直被投递。
失败重试机制:
当消费者出现异常后,消息会不断requeue(重新入队)到队列,再重新发送给消费者,然后再次异常,再次requeue无限循环,导致mg的消息处理飙升,带来不必要的压力。
我们可以利用spring的retry机制,在消费者出现异常时利用本地重试,而不是无限制的requeue到MQ队列
- spring:
- rabbitmq:
- listener:
- simple:
- prefetch: 1 #每次只能获取一条消息,处理完才能获取下一条消息
- acknowledge-mode: auto #none:关闭ack;manual:手动ack;auto:自动ack
- retry:
- enabled: true #开启消费者失败重试
- initial-interval: 1000ms #初始的失败等待时长为1秒
- multiplier: 1 #下次失败的等待时长倍数,下次等待时长 = multiplier * last-interval
- max-attempts: 3 #最大重试次数
- stateless: true #true无状态;false有状态。如果业务中包含事务,这里改为false
失败消息处理策略
在开启重试模式后,重试次数耗尽,如果消息依然失败,则需要有MessageRecoverer接口来处理,它包含三种不同的实现:
RejectAndDontRequeueRecoverer: 重试耗尽后,直接reject,丢弃消息。默认就是这种方式。ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队。RepublishMessageRecoverer: 重试耗尽后,将失败消息投递到指定的交换机。
RepublishMessageRecoverer原理
第一步、首先,定义接收失败消息的交换机、队列及其绑定关系。
第二步、然后,定义RepublishMessageRecoverer。
- @Slf4j
- @Configuration
- //@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry.enable",havingValue = "true") //开启了消费者失败重试机制才会生效
- public class ErrorConfiguration {
-
- @Bean
- public DirectExchange errorExchange(){
- return new DirectExchange("error.direct");
- }
-
- @Bean
- public Queue errorQueue(){
- return new Queue("error.queue");
- }
-
- @Bean
- public Binding errorBinding(Queue errorQueue,DirectExchange errorExchange){
- return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
- }
-
- @Bean
- public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){
- log.debug("加载RepublishMessageRecoverer");
- return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");
- }
- }
消费者如何保证消息一定被消费?
1、开启消费者确认机制为auto,由spring确认消息处理成功后返回ack,异常时返回nack。如果一直处理异常会一直重试,所以可以
2、开启消费者失败重试机制,并设置MessageRecoverer,多次重试失败后将消息投递到异常交换机,交由人工处理
幂等是一个数学概念,用函数表达式来描述是这样的: f(x)= f(f(X)。在程序开发中,则是指同一个业务,执行一次或多次对业务状态的影响是一致的。
方案一,是给每个消息都设置一个唯一id,利用id区分是否是重复消息
1、每一条消息都生成一个唯一的id,与消息一起投递给消费者。
2、消费者接收到消息后处理自己的业务,业务处理成功后将消息ID保存到数据库。(对业务有侵入)
3、如果下次又收到相同消息,去数据库查询判断是否存在,存在则为重复消息放弃处理。
消息发送方的启动类定义,生成唯一id,和消息一起投递
- @Bean
- public MessageConverter jacksonMessageConvertor() {
- //1.定义消息转换器
- Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
- //2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息
- jjmc.setCreateMessageIds(true);
- return jjmc;
- }
方案二,是结合业务逻辑,基于业务本身做判断。以我们的业务为例:我们要在支付后修改订单状态为已支付,应该在修改订单状态前先查询订单状态,判断状态是否是未支付。只有未支付订单才需要修改,其它状态不做处理。
方案三、使用Token令牌,生成一个token存储在redis中,请求的时候携带这个token一起请求(Token 最好将其放到 Headers 中),后端需要对这个Token作为 Key在redis中进行校验,如果 Key存在就执行删除命令,然后正常执行后面的业务逻辑。如果不存在对应的 Key 就返回重复执行的错误信息,这样来保证幂等操作。
首先,支付服务会正在用户支付成功以后利用MQ消息通知交易服务完成订单状态同步。
其次,为了保证MQ消息的可靠性,可以采用生产者确认机制、消费者确认、消费者失败重试等策略,确保消息投递和处理的可靠性。同时也开启了MQ的持久化,避免因服务宕机导致消息丢失。
最后,我们还在交易服务更新订单状态时做了业务幂等判断,避免因消息重复消费导致订单状态异常。
可以使用RepublishMessageRecoverer机制,进行人工介入。
我们可以在交易服务设置定时任务,定期查询订单支付状态。这样即便MQ通知失败,还可以利用定时任务作为兜底方案,确保订单支付状态的最终一致性。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。