【微服务系列笔记】MQ消息可靠性

本文涉及的产品
云原生网关 MSE Higress,422元/月
注册配置 MSE Nacos/ZooKeeper,118元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 消息可靠性涉及防止丢失,包括生产者发送时丢失、未到达队列以及消费者消费失败处理后丢失。确保RabbitMQ消息可靠性的方法有:开启生产者确认机制,确保消息到达队列;启用消息持久化以防止未消费时丢失;使用消费者确认机制,如设置为auto,由Spring确认处理成功后ack。此外,可开启消费者失败重试机制,多次失败后将消息投递到异常交换机。

3. 消息可靠性

消息丢失原因?

  • 发送时丢失:
  • 生产者发送的消息未送达exchange
  • 消息到达exchange后未到达queue
  • MQ宕机,queue将消息丢失
  • consumer接收到消息后未消费就宕机

如何确保RabbitMQ消息的可靠性?

  • 开启生产者确认机制,确保生产者的消息能到达队列
  • 开启持久化功能,确保消息未消费前在队列中不会丢失
  • 开启消费者确认机制为auto,由spring确认消息处理成功后完成ack
  • 开启消费者失败重试机制,并设置MessageRecoverer,多次重试失败后将消息投递到异常交换机,交由人工处理

3.1. 生产者确认机制

三种方式:

  1. 发送成功,返回ack publish-confirm
  • 调用ConfirmCallback,进行打印“消息发送成功, ID:{}”
  1. 发送到交换机,但没有到达队列,返回ack publish-return
  • 调用ConfirmCallback,进行打印“消息发送成功, ID:{}”
  • 调用ReturnCallback,进行日志打印“消息发送失败,应答码{},原因{},交换机{},路由键{},消息{}”,还可进行消息重发。
  1. 没有发送成功,返回 nack publish-confirm
  • 调用ConfirmCallback,进行打印“消息发送失败, ID:{}, 原因{}”

ConfirmCallback和ReturnCallback区别?

ConfirmCallback(发布确认回调):处理返回确认消息

ReturnCallback(发布返回回调):处理返回消息

回调函数理解,由于发送消息对于返回消息是异步回调


实现方式:修改配置

publish-confirm-type:开启publisher-confirm,这里支持两种类型:

  • simple:同步等待confirm结果,直到超时
  • correlated:异步回调,定义ConfirmCallback

publish-returns:回调机制,定义ReturnCallback

template.mandatory:定义消息路由失败时的策略。

  • true,则调用ReturnCallback
  • false:则直接丢弃消息
spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 1.消息体
    String message = "hello!";
    // 2.全局唯一的消息ID,需要封装到CorrelationData中
    CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
    // 3.添加callback
    cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
        @Override
        public void onSuccess(CorrelationData.Confirm result) {
            if (result.isAck()) {
                log.debug("消息投递到交换机成功, ID:{}", cd.getId());
            } else {
                log.error("消息投递到交换机失败,消息ID:{},原因:{}", cd.getId(), cd.getReturnedMessage());
            }
        }
        @Override
        public void onFailure(Throwable ex) {
            log.error("消息发送异常,ID:{},原因:{}", cd.getId(), ex.getMessage());
            rabbitTemplate.convertAndSend("amq.topic", routingKey, message, cd);
        }
    });
    //     cd.getFuture().addCallback(result -> {
    //         if (result.isAck()) {
    //             log.debug("消息投递到交换机成功, ID:{}", cd.getId());
    //         } else {
    //             log.error("消息投递到交换机失败,消息ID:{},原因:{}", cd.getId(), cd.getReturnedMessage());
    //         }
    //     }, ex -> log.error("消息发送异常,ID:{},原因:{}", cd.getId(), ex.getMessage()));
    // 4.发送消息
    rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);
    // 休眠一会儿,等待ack回执
    Thread.sleep(2000);
}
package cn.itcast.mq.config;
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("消息发送失败,应答码{},原因{},交换机{},路由键{},消息{}",
                     replyCode, replyText, exchange, routingKey, message.toString());
            // 如果有业务需要,可以重发消息
        });
    }
}

3.2. 消息持久化

交换机持久化,队列持久化,消息持久化

默认情况下三者都是持久化的,记住关键字durable,在MQ上看到属性feature带D

3.3. 消费者确认机制

manual

手动ack,需要在业务代码结束后,调用api发送ack。

none

关闭ack,MQ假定消费者获取消息后会成功处理,因此消息投递后立即被删除

消费者接收消息后,消费者异常,消息依然被RabbitMQ删除了。

auto

自动ack,由spring监测listener代码是否出现异常,没有异常则返回ack;抛出异常则返回nack

消费者接收消息后,消费者异常,消息会重入队,在重新发送给消费者,进行无限循环。

本地重试

  • 消息处理过程中抛出异常,不会重入队,而是在消费者本地重试
  • 重试达到最大次数后,执行失败策略

失败策略

  • RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认
  • ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队
  • RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机
spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000 # 初识的失败等待时长为1秒
          # 失败的等待时长倍数,下次等待时长 = multiplier * last-interval
          multiplier: 1 
          max-attempts: 3 # 最大重试次数
          # true无状态;false有状态。如果业务中包含事务,这里改为false
          stateless: true
package cn.itcast.mq.config;
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;
@Configuration
public class ErrorMessageConfig {
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }
    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}


相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
6月前
|
SQL 关系型数据库 数据库
【微服务系列笔记】Seata
Seata是一种开源的分布式事务解决方案,旨在解决分布式事务管理的挑战。它提供了高性能和高可靠性的分布式事务服务,支持XA、TCC、AT等多种事务模式,并提供了全局唯一的事务ID,以确保事务的一致性和隔离性。Seata还提供了分布式事务的协调、事务日志、事务恢复等功能,帮助开发人员简化分布式事务的管理和实现。
110 1
|
3月前
|
消息中间件 存储 Java
RabbitMQ 在微服务架构中的高级应用
【8月更文第28天】在微服务架构中,服务之间需要通过轻量级的通信机制进行交互。其中一种流行的解决方案是使用消息队列,如 RabbitMQ,来实现异步通信和解耦。本文将探讨如何利用 RabbitMQ 作为服务间通信的核心组件,并构建高效的事件驱动架构。
132 2
|
6月前
|
负载均衡 安全 Java
【微服务系列笔记】Gateway
Gateway是Spring Cloud生态系统中的网关服务,作为微服务架构的入口,提供路由、负载均衡、限流、鉴权等功能。借助于过滤器和路由器,Gateway能够动态地管理请求流量,保障系统的安全和性能。
157 7
|
2月前
|
消息中间件 JSON Java
|
3月前
|
消息中间件 Java RocketMQ
微服务架构师的福音:深度解析Spring Cloud RocketMQ,打造高可靠消息驱动系统的不二之选!
【8月更文挑战第29天】Spring Cloud RocketMQ结合了Spring Cloud生态与RocketMQ消息中间件的优势,简化了RocketMQ在微服务中的集成,使开发者能更专注业务逻辑。通过配置依赖和连接信息,可轻松搭建消息生产和消费流程,支持消息过滤、转换及分布式事务等功能,确保微服务间解耦的同时,提升了系统的稳定性和效率。掌握其应用,有助于构建复杂分布式系统。
65 0
|
3月前
|
消息中间件 存储 运维
RabbitMQ-消息消费时的可靠性保障
将这些实践融入到消息消费的处理逻辑中,可以很大程度上保障RabbitMQ中消息消费的可靠性,确保消息系统的稳定性和数据的一致性。这些措施的实施,需要在系统的设计和开发阶段充分考虑,以及在后续的维护过程中不断的调整和完善。
55 0
|
4月前
|
消息中间件 Kafka 索引
微服务数据问题之Broker宕机MetaQ保证数据的可靠性如何解决
微服务数据问题之Broker宕机MetaQ保证数据的可靠性如何解决
|
4月前
|
消息中间件 存储 缓存
架构设计篇问题之消息队列(MQ)在微服务系统中问题如何解决
架构设计篇问题之消息队列(MQ)在微服务系统中问题如何解决
|
6月前
|
消息中间件 存储 运维
深入理解MQ消息队列的高可用与可靠性策略
深入理解MQ消息队列的高可用与可靠性策略
1337 3
|
6月前
|
负载均衡 Java Apache
【微服务系列笔记】Feign
Feign是一个声明式的伪HTTP客户端,它使得HTTP请求变得更简单。使用Feign,只需要创建一个接口并注解。Feign默认集成了Ribbon,并和Eureka结合,默认实现了负载均衡的效果。 OpenFeign 是SpringCloud在Feign的基础上支持了SpringMVC的注解。
133 8