当前位置: 首页 > news >正文

RabbitMQ(六)消息的持久化

目录

    • 一、简介
      • 1.1 定义
      • 1.2 消息丢失的场景
    • 二、交换机的持久化
      • 方式一:直接 new
      • 方式二:channel.exchangeDeclare()
      • 方式三:ExchangeBuilder【推荐】
    • 三、队列的持久化
      • 方式一:直接 new
      • 方式二:channel.queueDeclare()
      • 方式三:QueueBuilder【推荐】
    • 四、消息的持久化
      • 方式一:channel.basicPublish()
      • 方式二:rabbitTemplate.convertAndSend()【推荐】
    • 五、持久化问题

在这里插入图片描述

一、简介

1.1 定义

  • 持久化: 是为了提高 RabbitMQ 消息的可靠性,防止在异常情况(重启宕机)下数据的丢失。
  • RabbitMQ 持久化分为三部分:交换机的持久化队列的持久化消息的持久化

1.2 消息丢失的场景

出现消息丢失的场景有:

  • 生产者发送消息丢失:发送消息到 RabbitMQ Server 异常。 可能因为网络问题造成 RabbitMQ 服务端无法收到消息。——解决方案:ConfirmCallback
  • 生产者发送消息丢失:消息无法路由到指定队列。 可能由于代码层面或配置层面错误导致消息路由到指定队列失败。——解决方案:ReturnCallback
  • RabbitMQ Server 存储消息丢失:消息未完全持久化到磁盘。 可能因为 RabbitMQ 宕机导致消息未完全持久化,或队列丢失从而导致消息丢失等持久化问题。——解决方案:集群部署,实现高可用
  • 消费者消费消息丢失:消费者消费消息异常。 可能在消费者接收消息后,还没来得及消费消息,消费者就发生宕机、故障等问题。——解决方案:消费端手动确认消息

二、交换机的持久化

方式一:直接 new

直接实例化对应的 Exchange 实现类即可,默认:持久化的、非自动删除的

@Bean
public DirectExchange directExchange() {return new DirectExchange(directExchange);
}

Exchange 源码:

package org.springframework.amqp.core;public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {...public AbstractExchange(String name) {// 默认是持久化的、非自动删除的。this(name, true, false);}...
}

方式二:channel.exchangeDeclare()

在 RabbitMQ 的原生 API 中,例如:Java 客户端 API,声明持久化交换机时,需要将 durable 参数设置为 true

// 声明交换机:(交换机名称,交换机类型,持久化)
channel.exchangeDeclare("myExchange", "direct", true);

方式三:ExchangeBuilder【推荐】

在 Spring AMQP 中,我们可以使用 ExchangeBuilder 来创建持久化的交换机:

import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;Exchange exchange = ExchangeBuilder.directExchange("myExchange").durable(true) // 交换机持久化.build();

三、队列的持久化

方式一:直接 new

直接实例化 Queue 类即可,默认:持久化的、非独占的、非自动删除的

@Bean
public Queue myQueue() {return new Queue("myQueue");
}

Queue 源码:

package org.springframework.amqp.core;public Queue(String name) {// 默认是持久化的、非独占的、非自动删除的。this(name, true, false, false);
}

方式二:channel.queueDeclare()

在原生 RabbitMQ API 中,例如 Java 客户端 API,声明持久化队列时,需要将 durable 参数设置为 true

// 声明队列:(队列名称,持久化,非独占,非自动删除,可选队列参数)
channel.queueDeclare("myQueue", true, false, false, null);

方式三:QueueBuilder【推荐】

在 Spring AMQP 中,可以使用 QueueBuilder 来创建持久化的队列:

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;Queue queue = QueueBuilder.durable("myQueue").build();

四、消息的持久化

即使队列时持久化的,发送到队列中的消息默认情况下 可能仍然是非持久化的。要使消息持久化,需要在发送消息时设置消息属性为持久化。

方式一:channel.basicPublish()

在原生 RabbitMQ API 中,例如 Java 客户端 API,声明持久化消息时,需要将 deliveryMode 参数设置为 2

import com.rabbitmq.client.AMQP;AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().deliveryMode(2) // 这里2对应MessageDeliveryMode.PERSISTENT.build();// 参数:(交换机,路由键,消息的其他参数,消息体)
channel.basicPublish("myExchange", "myRoutingKey", props. messageBytes);

或者,我们可以直接使用 MessageProperties.PERSISTENT_TEXT_PLAIN 来进行指定消息的持久化:

import com.rabbitmq.client.MessageProperties;// 参数:(交换机,路由键,消息的其他参数,消息体)
channel.basicPublish("myExchange", "myRoutingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

方式二:rabbitTemplate.convertAndSend()【推荐】

在 Spring AMQP 中,可以使用 rabbitTemplate.convertAndSend() 来创建持久化的消息:

rabbitTemplate.convertAndSend("myQueue", body, message -> {message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;
});

通过上述步骤,消息会被持久化存储,只要队列也被正确配置为持久化,即使 RabbitMQ 服务器重启,消息也将被保留下来。

补充:

然而,即使队列和消息都是持久化的,也不能完全保证消息的 100% 不丢失。例如:在消息尚未被刷写到磁盘时,RabbitMQ 服务器突然崩溃,这种情况下的消息仍然可能会丢失。此外,RabbitMQ 不保证消息的立即持久化,而是尽可能快地将消息保存到磁盘


五、持久化问题

思考:将交换机、队列、消息都设置持久化之后就能保证数据不会丢失吗?

答:当然不能,需要从多方面考虑:

  • 场景1: 如果消费者订阅队列的时候将 autoAck(自动确认)设置为 true,虽然消费者接收到了消息,但是没有来得及处理就宕机了,那消息也会丢失。

    解决方案: 将消费者的消息确认机制改为手动确认,带处理完消息之后,手动删除消息。

  • 场景2: 在 RabbitMQ 服务器,如果消息正确被发送,但是 RabbitMQ 服务器中的消息还没来得及持久化,即没有将数据写入磁盘时,如果服务器发生异常,消息也会丢失。

    解决方案: 可以 通过 RabbitMQ 集群的方式实现消息中间件的高可用

因此,还需要做好生产者和消费者的 消息确认机制 以及通过 集群 的方式来实现 RabbitMQ 的高可用。

整理完毕,完结撒花~ 🌻





参考地址:

1.RabbitMQ保证消息可靠性,https://www.cnblogs.com/auguse/articles/17712620.html

http://www.lryc.cn/news/279744.html

相关文章:

  • mybatisplus配置
  • node-sass@4.7.2 postinstall: `node scripts/build.js`
  • pyDAL一个python的ORM(9) pyDAL的嵌套查询
  • 融资项目——EasyExcel操作Excel文件
  • OAI openair3-NAS-UE-EMM代码解读(续)
  • SQL-条件查询与聚合函数的使用
  • Qt移植曲线显示
  • 基于springboot生鲜交易系统源码和论文
  • vue中鼠标拖动触发滚动条的移动
  • 1、理解Transformer:革新自然语言处理的模型
  • Acwing847 图中点的层次(bfs)
  • windows11通过虚拟机安装Ubuntu20.04
  • 时序预测 | Matlab实现EEMD-SSA-BiLSTM、EEMD-BiLSTM、SSA-BiLSTM、BiLSTM时序预测对比
  • Android14之解决Pixel手机联网出现感叹号(一百八十)
  • Vmware虚拟机问题解决方案 运行虚拟机系统蓝屏 运行虚拟机时报错VT-x
  • uni-app中轮播图实现大图预览
  • 了解什么是UV纹理?
  • 【蓝桥备赛】wzy的数组Ⅱ——简单莫队问题
  • 学习Qt笔记
  • pymssql 报错误解决办法:20002, severity 9
  • Web缓存代理
  • Rust-模式解构
  • C#基于ScottPlot进行可视化
  • 基于JAVA+ssm开发的在线报名系统设计与实现【附源码】
  • 蓝桥——第 3 场 小白入门赛(A-D)
  • Java项目:06 Springboot的进销存管理系统
  • 数据结构与算法之美学习笔记:47 | 向量空间:如何实现一个简单的音乐推荐系统?
  • 5《Linux》
  • go-carbon v2.3.5 发布,轻量级、语义化、对开发者友好的 golang 时间处理库
  • VQ-VAE(Neural Discrete Representation Learning)论文解读及实现