首页 > 其他分享 >rabbitmq消息队列之持久化

rabbitmq消息队列之持久化

时间:2023-04-03 17:01:03浏览次数:34  
标签:持久 String exchange 队列 param rabbitmq 消息

在生产过程中,难免会发生服务器宕机的事情,RabbitMQ也不例外,可能由于某种特殊情况下的异常而导致RabbitMQ宕机从而重启,那么这个时候对于消息队列里的数据,包括交换机、队列以及队列中存在消息恢复就显得尤为重要了。RabbitMQ本身带有持久化机制,包括交换机、队列以及消息的持久化。持久化的主要机制就是将信息写入磁盘,当RabbtiMQ服务宕机重启后,从磁盘中读取存入的持久化信息,恢复数据。

1、交换机的持久化

如果使用常规的声明交换机的方法:

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

默认不是持久化的,在服务器重启之后,交换机会消失。我们在管理台的Exchange页签下查看交换机,可以看到使用上述方法声明的交换机,Features一列是空的,即没有任何附加属性。

我们换用另一种方法声明交换机:

channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true);

查看一下方法的说明:

 * Actively declare a non-autodelete exchange with no extra arguments
 * @see com.rabbitmq.client.AMQP.Exchange.Declare
 * @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
 * @param exchange the name of the exchange
 * @param type the exchange type
 * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
 * @throws java.io.IOException if an error is encountered
 * @return a declaration-confirm method to indicate the exchange was successfully declared
 */
Exchange.DeclareOk (String exchange, String type, boolean durable) throws IOException;

我们可以看到第三个参数durable,如果为true时则表示要做持久化,当服务重启时,交换机依然存在.

   

2、队列的持久化

与交换机的持久化相同,队列的持久化也是通过durable参数实现的,看一下方法的定义:

 * Declare a queue
 * @see com.rabbitmq.client.AMQP.Queue.Declare
 * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
 * @param queue the name of the queue
 * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
 * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
 * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
 * @param arguments other properties (construction arguments) for the queue
 * @return a declaration-confirm method to indicate the queue was successfully declared
 * @throws java.io.IOException if an error is encountered
 */
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;

第二个参数跟交换机方法的参数一样,true表示做持久化,当RabbitMQ服务重启时,队列依然存在。

这里说一下另外两个参数:

  • exclusive:排他队列。如果一个队列被声明为排他队列,那么这个队列只能被第一次声明他的连接所见,并在连接断开的时候自动删除。这里有三点需要说明:
    • 排他队列是基于连接可见的,同一连接的不同信道是可以同时访问同一连接创建的排他队列
    • 如果一个连接已经声明了一个排他队列,其他连接是不允许建立同名的排他队列的,这个与普通队列不同
    • 即使该队列是持久化的,一旦连接关闭或者客户端退出,该排他队列都会被自动删除的,这种队列适用于一个客户端发送读取消息的应用场景
  • autoDelete:自动删除,如果该队列没有任何订阅的消费者的话,该队列会被自动删除。这种队列适用于临时队列

3、消息的持久化

消息的持久化是指当消息从交换机发送到队列之后,被消费者消费之前,服务器突然宕机重启,消息仍然存在。消息持久化的前提是队列持久化,假如队列不是持久化,那么消息的持久化毫无意义。

通过如下代码设置消息的持久化:

channel.basicPublish(EXCHANGE_NAME,"",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

其中MessageProperties.PERSISTENT_TEXT_PLAIN是设置持久化的参数。

看一下basicPublish方法的定义:

 * Publish a message
 * @see com.rabbitmq.client.AMQP.Basic.Publish
 * @param exchange the exchange to publish the message to
 * @param routingKey the routing key
 * @param props other properties for the message - routing headers etc
 * @param body the message body
 * @throws java.io.IOException if an error is encountered
 */
void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;

再看下BasicProperties的类型:

public BasicProperties(
    String contentType,
    String contentEncoding,
    Map<String,Object> headers,
    Integer deliveryMode,
    Integer priority,
    String correlationId,
    String replyTo,
    String expiration,
    String messageId,
    Date timestamp,
    String type,
    String userId,
    String appId,
    String clusterId)

其中deliveryMode是设置消息持久化的参数,等于1不设置持久化,等于2设置持久化。PERSISTENT_TEXT_PLAIN是实例化的一个deliveryMode=2的对象,便于编程:

public static final BasicProperties PERSISTENT_TEXT_PLAIN =
    new BasicProperties("text/plain",
                        null,
                        null,
                        2,
                        0, null, null, null,
                        null, null, null, null,
                        null, null);

保证在服务器重启的时候可以保持不丢失相关信息,重点解决服务器的异常崩溃而导致的消息丢失问题。但是,将所有的消息都设置为持久化,会严重影响RabbitMQ的性能,写入硬盘的速度比写入内存的速度慢的不只一点点。对于可靠性不是那么高的消息可以不采用持久化处理以提高整体的吞吐率,在选择是否要将消息持久化时,需要在可靠性和吞吐量之间做一个权衡。

4、进一步思考

将Exchange、Queue、Message都设置了持久化之后就能保证100%保证数据不丢失了吗?

答案是否定的。

首先,从consumer端来说,如果这时autoAck=true,那么当consumer接收到相关消息之后,还没来得及处理就crash掉了,那么这样也算数据丢失,这种情况也好处理,只需将autoAck设置为false,然后在正确处理完消息之后进行手动ack。

还有要在producer引入事务机制或者Confirm机制来确保消息已经正确的发送至broker端。

其次,关键的问题是消息在正确存入RabbitMQ之后,还需要有一段时间(这个时间很短,但不可忽视)才能存入磁盘之中,RabbitMQ并不是为每条消息都做fsync的处理,可能仅仅保存到cache中而不是物理磁盘上,在这段时间内RabbitMQ broker发生crash,消息保存到cache但是还没来得及落盘,那么这些消息将会丢失。

那么这个怎么解决呢?首先可以引入RabbitMQ的镜像队列,这个相当于配置了副本,当master在此特殊时间内crash掉,可以自动切换到slave,这样有效的保障了HA,除非整个集群都挂掉,这样也不能完全的100%保障RabbitMQ不丢消息,但比没有镜像队列的要好很多,很多现实生产环境下都是配置了镜像队列的。

RabbitMQ的可靠性除了Exchange、Queue、Message的持久化,还涉及producer端的确认机制、broker端的镜像队列的配置以及consumer端的确认机制,要想确保消息的可靠性越高,那么性能也会随之而降,鱼和熊掌不可兼得,关键在于选择和取舍。



作者:冰河winner
链接:https://www.jianshu.com/p/84b3e5d9f8f8
来源:简书
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。

标签:持久,String,exchange,队列,param,rabbitmq,消息
From: https://www.cnblogs.com/zhanchenjin/p/17283600.html

相关文章

  • RabbitMQ在本地环境下不消费数据
    packagecom.test.core.condition;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.stereoty......
  • 详细解析Java异步线程处理队列任务工具类以及实战
    场景待入快速理解小场景描述:【一群人】来到【一个大厅】办理业务,大厅中有【多个窗口】给我们办理业务。每个人都有自己要办事情,处理过程需要消耗时间。大厅根据人群多少,开始窗口梳理。如果把“一群人”理解成一群待处理的n个【任务】,把这群人排成一个长队就形成了一个【任......
  • 数据结构 第三章 栈与队列
    之前期末考试,大部分都是二叉树,先根遍历之类的,还有一些辨析题目,一些很零碎的知识点,关于二叉树,这些的栈1.栈的概念首先对于线性表来说,线性表的插入和删除操作可以在任意的位置进行,而栈的插入和删除操作只允许在表的尾端进行。栈中,允许进行插入和删除操作的一端称为栈顶,另一端称......
  • RabbitMQ 07 发布订阅模式
    发布订阅模式发布订阅模式结构图:比如信用卡还款日临近了,那么就会给手机、邮箱发送消息,提示需要去还款了,但是手机短信和邮件发送并不一定是同一个业务提供的,但是现在又希望能够都去执行,就可以用到发布订阅模式,简而言之就是,发布一次,消费多个。实现这种模式需要用到另一种类型的......
  • K8S用yaml资源清单部署redis数据库,数据持久化保存
    1.创建redis存储数据目录和配置文件mkdir-p/data/redis/redis-config/touch/data/redis/redis-config/redis.confcat>>/data/redis/redis-config/redis.conf<<EOFrequirepass123456bind0.0.0.0save9001save30010save6010000EOF2.编写redis的yaml资源清单api......
  • RabbitMQ下载与安装
    1.首先进入rabbitmq官网找,查看rabbitmq对应的erlang之间对应的版本信息网址:RabbitMQErlangVersionRequirements—RabbitMQ网站看不懂的话可以使用windows系统最新的edge浏览器,有翻译功能.2.在github网站下载erlang的.npm文件(文件下载可能较慢,耐心等待)网址:http......
  • Docker安装RabbitMQ
     安装步骤:一般情况,需要先安装部署Erlang环境再安装RabbitMQ环境。不过docker中,Erlang环境已经被docker自动安装,我们只需要简单几步操作即可完成RabbitMQ环境地安装+Web页面管理。端口开放如果在云服务上部署需在安全组开通一下端口:15672、5672、25672、61613、1883。15672(UI......
  • day10| 225.用队列实现栈;232.用栈实现队列
    题目简述: 请你仅使用两个队列实现一个后入先出(LIFO)的栈,并支持普通栈的全部四种操作(push、top、pop和empty)。实现MyStack类:voidpush(intx)将元素x压入栈顶。intpop()移除并返回栈顶元素。inttop()返回栈顶元素。booleanempty()如果栈是空的,返回true;否则,返回......
  • 2023-04-01-循环队列CycleSqQueue的基本操作
    //循环链表//牺牲一个单元来区分队空还是队满#include<stdio.h>#include<stdbool.h>#defineMAXSIZE6typedefstruct{intdata[MAXSIZE];intfront,rear;}CySqQueue;voidinitCySqQueue(CySqQueue*C)//初始化循环链表{C->front=0;C->rear=0;......
  • 开源优先队列FastPriorityQueue源码阅读
    FastPriorityQueue  源码连接:https://github.com/BlueRaja/High-Speed-Priority-Queue-for-C-Sharp  大致结构:  1节点在内存中的结构还是数组,且首节点为无意义节点,有效节点从索引1开始。(见FastPriorityQueue的T[]_nodes)  2维护的节点必须时固定的继承。(见FastPri......