RabbitMQ之生产者可靠性

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

1.生产者重连

有的时候由于网络波动,可能会出现客户端连接RabbitMQ失败的情况。通过配置我们可以开启连接失败后的重连机制

# Spring配置信息
spring:
  # Rabbitmq配置
  rabbitmq:
    # 设置RabbitMQ连接超时时间
    connection-timeout: 2s
    template:
      retry:
        # 开启超时重试机制
        enabled: true
        # 失败后的初始等待时间
        initial-interval: 1000ms
        # 失败后下次的等待时长倍数,下次等待时间 = initial-interval * multiplier
        multiplier: 1
        # 最大重试次数
        max-attempts: 3

注:当网络不稳定的时候,利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试,也就是说多次重试等待的过程中,当前线程是被阻塞的,会影响业务性能。如果对于业务性能有要求,建议禁用重试机制,如果一定要使用,需合理配置等待时长和重试次数,也可以考虑使用异步线程来执行发送消息的代码

2.生产者确认

RabbitMQ提供了Publisher Confirm和Publisher Return两种确认机制。开启确认机制后,在RabbitMQ成功收到消息后会返回确认消息给生产者

(1)RabbitMQ返回的结果情况

  • 消息投递到了RabbitMQ,但是路由失败。此时会通过Publisher Return返回路由异常原因,然后返回ACK,告知投递成功(造成原因:代码有问题或者交换机的配置有问题)
  • 临时消息(非持久化)投递到了RabbitMQ,并且成功入队,返回ACK,告知投递成功
  • 持久消息投递到了RabbitMQ,并且入队完成持久化,返回ACK,告知投递成功
  • 其他情况都会返回NACK,告知投递失败

(2)生产者确认代码实现

添加配置文件信息

# Spring配置信息
spring:
  # Rabbitmq配置
  rabbitmq:
    # 开启publisher confirm机制 none:关闭publisher confirm机制 simple:同步阻塞等待MQ的回执消息 correlated:MQ异步回调方式返回回执消息
    publisher-confirm-type: correlated
    # 开启publisher return机制
    publisher-returns: true

编写Publisher Return回调函数(注:每个RabbitTemplate只能配置一个ReturnCallback)

import lombok.extern.slf4j.Slf4j;
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((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息返回:{},{},{},{},{}",
                    message, replyCode, replyText, exchange, routingKey);
        });
    }
}

编写ConfirmCallback(注:在每一个消息发送时候单独指定)

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 org.springframework.util.concurrent.ListenableFutureCallback;
@Slf4j
@Component
public class Producer {
    @Autowired
    RabbitTemplate rabbitTemplate;
    public void sendMessage(Object message){
        // 1.创建CorrelationData对象
        CorrelationData correlationData = new CorrelationData(randomUUID().toString());
        // 2.给Future添加ConfirmCallback
        correlationData.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                // Future发生异常时的处理逻辑,基本不会触发
                log.error("handle message ack fail",ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                // Future接收到回执的处理逻辑,参数中的result就是回执内容
                if(result.isAck()){ // result.isAck(),boolean类型,true代表收到ack,false代表收到nack
                    log.debug("发送消息成功,收到 ack!");
                }else{// result.getReason(),String类型,返回nack时的异常原因
                    log.error("发送消息失败,收到 nack,reason : {}" , result.getReason());
                }
            }
        });
        // 发送消息
        rabbitTemplate.convertAndSend("交换机名称", "routingKey值", message,correlationData);
    }

}

3.如何处理生产者的确认消息

(1)生产者确认需要额外的网络和系统资源开销,尽量不要使用
(2)如果一定要使用,无需开启Publisher-Return机制,因为一般路由失败是自己的业务问题
(3)对于nack消息可以有限次数重试,依然失败则记录异常信息文章来源地址https://www.toymoban.com/news/detail-815877.html

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

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

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

相关文章

  • 【RabbitMQ】RabbitMQ 消息的可靠性 —— 生产者和消费者消息的确认,消息的持久化以及消费失败的重试机制_rabbitmq 生产者消息确认

    先自我介绍一下,小编浙江大学毕业,去过华为、字节跳动等大厂,目前阿里P7 深知大多数程序员,想要提升技能,往往是自己摸索成长,但自己不成体系的自学效果低效又漫长,而且极易碰到天花板技术停滞不前! 因此收集整理了一份《2024年最新大数据全套学习资料》,

    2024年04月26日
    浏览(85)
  • 【RabbitMQ】RabbitMQ 消息的可靠性 —— 生产者和消费者消息的确认,消息的持久化以及消费失败的重试机制

    在现代分布式应用程序中,消息队列扮演了至关重要的角色,允许系统中的各个组件之间进行异步通信。这种通信模式提供了高度的灵活性和可伸缩性,但也引入了一系列的挑战,其中最重要的之一是消息的可靠性。 首先让我们来了解一下,在消息队列中,消息从生产者发送

    2024年02月05日
    浏览(49)
  • RabbitMQ --- 消息可靠性

    消息队列在使用过程中,面临着很多实际问题需要思考:      消息从发送,到消费者接收,会经理多个过程: 其中的每一步都可能导致消息丢失,常见的丢失原因包括: 发送时丢失: 生产者发送的消息未送达exchange 消息到达exchange后未到达queue MQ宕机,queue将消息丢失 co

    2024年02月14日
    浏览(46)
  • rabbitmq笔记-rabbitmq进阶-数据可靠性,rabbitmq高级特性

    消息何去何从 mandatory和immediate是channel.basicPublish方法的两个参数,都有消息传递过程中不可达目的地时将消息返回给生产者的功能。 mandatory参数 true:交换器无法根据自身的类型 和路由键找到符合条件的队列,rabbitmq调用Basic.Return命令将消息返回给生产者 生产者调用channel.

    2024年02月10日
    浏览(43)
  • RabbitMQ消息的可靠性

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

    2024年04月14日
    浏览(70)
  • RabbitMQ-保证消息可靠性

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

    2024年02月07日
    浏览(48)
  • RabbitMQ保证消息的可靠性

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

    2024年02月19日
    浏览(46)
  • 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日
    浏览(48)
  • RabbitMQ之MQ可靠性

    RabbitMQ实现数据持久化包括3个方面 (1)交换机持久化 (2)队列持久化 (3)消息持久化 注:开启持久化和生产者确认时,RabbitMQ只有在消息持久化完成后才会给生产者返回ACK回执 从RabbitMQ的3.6.0版本开始,就增加了Lazy Queue的概念,也就是惰性队列 注:从3.12版本后,所有队

    2024年01月21日
    浏览(37)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包