RabbitMQ消息可靠性问题及解决

这篇具有很好参考价值的文章主要介绍了RabbitMQ消息可靠性问题及解决。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

说明:在RabbitMQ消息传递过程中,有以下问题:

  • 消息没发到交换机

  • 消息没发到队列

  • MQ宕机,消息在队列中丢失

  • 消息者接收到消息后,未能正常消费(程序报错),此时消息已在队列中移除

针对以上问题,提供以下解决方案:

  • 消息确认:确认消息是否发送到交换机、队列;

  • 消息持久化:持久化消息,以防MQ宕机造成消息丢失;

  • 消费者消息确认:确认消费者已正确消费消息,才把消息从队列中删除;

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

消息确认

可以使用Rabbit MQ提供的publisher confirm机制来避免消息发送到MQ过程丢失。具体实现是,publisher-confirm(发送者确定)、publisher-return(发送者回执),前者判断消息到交换机、后者判断交换机到队列


publisher-confirm(发送者确定)

  • 消息成功投递到交换机,返回ack;

  • 消息未投递到交换机,返回nack;

publisher-return(发送者回执)

  • 消息投递到交换机,但没有到队列,返回ack,即失败原因;

在生产者端添加配置

spring:
  rabbitmq:
    # rabbitMQ相关配置
    host: 118.178.228.175
    port: 5672
    username: root
    password: 123456
    virtual-host: /

    # 开启生产者确认,correlated为异步,simple为同步
    publisher-confirm-type: correlated

    # 开启publish-return功能,基于callback机制
    publisher-returns: true

    # 开启消息路由失败的策略,true是调用returnCallback方法,false是丢弃消息
    template:
      mandatory: true

publisher-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 {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 获取RabbitTemplate对象
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);

        // 设置ReturnCallback
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {

            /**
             * 回执信息
             * @param message 信息对象
             * @param replyCode 回执码
             * @param replyText 回执内容
             * @param exchange 交换机
             * @param routingKey 路由键值
             */
            @Override
            public 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(发送者确定)代码

    @Test
    public void sendExceptionMessage() {
        // 路由键值
        String routingKey = "exception";

        // 消息
        String message = "This is a exception message";

        // 给消息设置一个唯一ID
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        // 编写confirmCallBack回调函数
        correlationData.getFuture().addCallback(new SuccessCallback<CorrelationData.Confirm>() {
            @Override
            public void onSuccess(CorrelationData.Confirm confirm) {
                if (confirm.isAck()) {
                    // 消息发送交换机成功
                    log.debug("消息送达至交换机成功");
                } else {
                    // 消息发送交换机失败,打印消息
                    log.error("消息未能送达至交换机,ID{},原因{}", correlationData.getId(), confirm.getReason());
                }
            }
        }, new FailureCallback() {
            // 消息发送交换机异常
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息发送交换机异常,ID:{},原因{}", correlationData.getId(), ex.getMessage());
            }
        });

        rabbitTemplate.convertAndSend("amq.direct", routingKey, message, correlationData);
    }

测试,设置一个不存在的routingKey,被发送者确认(publisher-confirm)捕获到;

// 路由键值
String routingKey = "null";

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

设置一个不存在的路由,被发送者回执(publisher-return)捕获到;

rabbitTemplate.convertAndSend("null", routingKey, message, correlationData);

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

消息持久化

消息持久化,是指把消息保存到磁盘中,在RabbitMQ宕机或者关机时,重启后,消息仍可以保存下来。消息依赖于交换机、队列,因此持久化消息,同时也需要持久化交换机、队列。

创建一个持久化的交换机、队列

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 消息持久化
 */
@Configuration
public class DurableConfig {

    /**
     * 交换机持久化
     * @return
     */
    @Bean
    public DirectExchange directExchange(){
        // 三个参数分别是:交换机名、是否持久化、没有队列与之绑定时是否自动删除
        return new DirectExchange("durable.direct",true,false);
    }

    /**
     * 队列持久化
     * @return
     */
    @Bean
    public Queue durableQueue(){
        return QueueBuilder.durable("durable.queue").build();
    }

    /**
     * 交换机与队列绑定
     * @return
     */
    @Bean
    public Binding binding(){
        return BindingBuilder.bind(durableQueue()).to(directExchange()).with("durable");
    }

}

发送一个持久化的消息

    /**
     * 发送持久化消息
     */
    @Test
    public 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",表示该消息是持久化消息

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

(源码:MessageDeliveryMode类)
RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

实际上,交换机、队列默认就是持久化的(durable: true),所以不用特意设置;

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

消费者消息确认

介绍

消费者消息确认,是为了确保消费者已经消费了消息,才让MQ把该消息删除;

可通过在消费者的配置文件中增加下面这行配置实现,备选项有以下三个:

  • none:关闭ack,表示不做处理,消息发给消费者之后就立即被删除;

  • auto:自动ack,表示由Spring检测代码是否出现异常,出现异常则保留消息,没有异常则删除消息;

  • manual:手动ack,可根据业务手动编写代码,返回ack;

spring:
  rabbitmq:
    listener:
      simple:
      	# 设置消息确认模式
        acknowledge-mode: none

测试:none

可编写代码测试,下面是生产者代码,发送消息

    /**
     * 发送普通消息
     */
    @Test
    public 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);
    }

测试结果,程序报错,消息也没能保留下来

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java
RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

测试:auto

更改设置为:auto,重试

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

但是消息未被删除

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

这种情况,在实际开发中是不能允许,可以通过更改消费失败的重试机制解决。

消费失败重试机制

方法一:设置retry

因为消息被消费失败,消息会一直循环重试,无限循环,导致mq的消息处理飙升,带来不必要的压力,这种情况可以通过在消费者端添加以下配置,限制失败重试的条件来解决:

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          # 开启消费者失败重试
          enabled: true
          # 初次失败等待时长为1秒
          initial-interval: 1000
          # 失败的等待时长倍数,即后一次等待的时间是前一次等待时间的多少倍
          multiplier: 1
          # 最多重试次数
          max-attempts: 3
          # true 无状态 false 有状态 如果业务中包含事务 改为false
          stateless: true

开启后,控制台可以发现,信息不回一直循环打印,而是打印数条后停止,日志信息中有提示“Retry Policy Exhausted”(重试策略已用尽)

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java
这种通过配置的方式,并不会重试数次后仍保留消息,而是重试数次仍失败,随即丢弃消息,消息丢失,这在实际开发中也是不能被允许的。

方法二:路由存储消息

因此,可以通过下面这个方法,把消费失败的消息,通过交换机路由到另外的队列中存储起来,等业务代码被修复,再路由回来消费。

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

代码如下

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
     */
    @Bean
    public DirectExchange errorExchange(){
        return new DirectExchange("error.direct");
    }

    /**
     * 创建一个队列,用于存储消费失败的消息
     * @return
     */
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue");
    }

    /**
     * 绑定
     * @return
     */
    @Bean
    public Binding errorBinding(){
        return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");
    }

    /**
     * 路由,当消费失败时,把消费失败的消息路由到此队列中,路由key为"error"
     * @param rabbitTemplate
     * @return
     */
    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");
    }
}

可以看到,消息消费失败后并没有被丢失,而是路由到错误队列中存储了起来。因为错误队列没有设置RabbitListener,所以可以存储消息,等带代码问题被排查出来后,可以再针对该队列设置监听方法,消费这部分错误的消息。

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

另外,值得一提的是,消费者这边的控制台会报一个警告,提示路由密钥错误。我们可以理解,在RabbitMQ底层,会把消费失败了的消息,统一路由到一个地方去,而我们这种手动把消费失败的消息路由到自定义的队列中的方式,打破了这种“默认的规则”,所以报了一个这样的警告。这种警告是在可控范围内的。

RabbitMQ消息可靠性问题及解决,rabbitmq,分布式,java

总结

RabbitMQ发送消息,为了确保消息的可靠性,保证消息能被交换机、队列收到,消息能被正常消费,而不会因消费失败而丢失,提供了对应的一系列方法,并且最后还提供了两种消费失败重试方法,优化了消费过程,非常Nice。文章来源地址https://www.toymoban.com/news/detail-601451.html

到了这里,关于RabbitMQ消息可靠性问题及解决的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • RabbitMQ消息的可靠性

    面试题: Rabbitmq怎么保证消息的可靠性? 1.消费端消息可靠性保证: 消息确认(Acknowledgements) : 消费者在接收到消息后,默认情况下RabbitMQ会自动确认消息(autoAck=true)。为保证消息可靠性,可以设置autoAck=false,使得消费者在处理完消息后手动发送确认(basicAck)。如果消费

    2024年04月14日
    浏览(69)
  • RabbitMQ如何保证消息可靠性

    目录 1、RabbitMQ消息丢失的可能性 1.1 生产者消息丢失场景 1.2 MQ导致消息丢失 1.3 消费者丢失 2、如何保证生产者消息的可靠性 2.1 生产者重试机制 2.2 生产者确认机制 2.3 实现生产者确认 2.3.1 配置yml开启生产者确认 2.3.2 定义ReturnCallback 2.3.3 定义ConfirmCallback 3、MQ消息可靠性 3.1

    2024年02月20日
    浏览(51)
  • RabbitMQ高级篇---消息可靠性

    1、消息可靠性: 消息从发送到消费者接受,会经历多个过程,每个消息传递的过程都可能导致消息的丢失: 常见的丢失原因: 发送时消息丢失原因: 生产者发送的消息未送达exchange 消息到达exchange后未到达queue MQ宕机,queue将消息丢失 consumer接收到消息后未消费就宕机 Rab

    2024年01月20日
    浏览(47)
  • RabbitMQ保证消息的可靠性

    消息从发送,到消费者接收,会经理多个过程: 其中的每一步都可能导致消息丢失,常见的丢失原因包括: 发送时丢失: 生产者发送的消息未送达exchange 消息到达exchange后未到达queue MQ宕机,queue将消息丢失 consumer接收到消息后未消费就宕机 针对这些问题,RabbitMQ分别给出了

    2024年02月19日
    浏览(46)
  • rabbitmq消息可靠性之消息回调机制

    rabbitmq消息可靠性之消息回调机制 rabbitmq在消息的发送与接收中,会经过上面的流程,这些流程中每一步都有可能导致消息丢失,或者消费失败甚至直接是服务器宕机等,这是我们服务接受不了的,为了保证消息的可靠性,rabbitmq提供了以下几种机制 生产者确认机制 消息持久

    2024年02月08日
    浏览(52)
  • 【RabbitMQ】之消息的可靠性方案

    一、数据丢失场景 二、数据可靠性方案 1、生产者丢失消息解决方案 2、MQ 队列丢失消息解决方案 3、消费者丢失消息解决方案 MQ 消息数据完整的链路为 :从 Producer 发送消息到 RabbitMQ 服务器中,再由 Broker 服务的 Exchange 根据 Routing_Key 路由到指定的 Queue 队列中,最后投送到消

    2024年02月14日
    浏览(40)
  • RabbitMQ之消息的可靠性传递

    提示:这里可以添加系列文章的所有文章的目录,目录需要自己手动添加 RabbitMQ之消息的可靠性传递 提示:写完文章后,目录可以自动生成,如何生成可参考右边的帮助文档 提示:这里可以添加本文要记录的大概内容: 在当今的信息化时代,消息传递在企业级应用和分布式

    2024年01月19日
    浏览(41)
  • rabbitmq如何保证消息的可靠性

    RabbitMQ可以通过以下方式来保证消息的可靠性: 在发布消息时,可以设置消息的delivery mode为2,这样消息会被持久化存储在磁盘上,即使RabbitMQ服务器重启,消息也不会丢失。 可以创建持久化的队列,这样即使RabbitMQ服务器重启,队列也不会丢失。 在消费者端,可以 设置手动

    2024年01月23日
    浏览(52)
  • RabbitMQ 能保证消息可靠性吗

    手把手教你,本地RabbitMQ服务搭建(windows) 消息队列选型——为什么选择RabbitMQ RabbitMQ灵活运用,怎么理解五种消息模型 推或拉? RabbitMQ 消费模式该如何选择 死信是什么,如何运用RabbitMQ的死信机制? 前面我们在做MQ组件选型时,提到了rabbitMQ的消息可靠性,那么它到底可靠

    2024年02月16日
    浏览(46)
  • Rabbitmq怎么保证消息的可靠性?

    一、消费端消息可靠性保证 : 消息确认(Acknowledgements) : 消费者在接收到消息后,默认情况下RabbitMQ会自动确认消息(autoAck=true)。为保证消息可靠性,可以设置autoAck=false,使得消费者在处理完消息后手动发送确认(basicAck)。如果消费者在处理过程中发生异常或者未完成

    2024年04月14日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包