java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > RabbitMQ持久化机制

RabbitMQ消息队列之持久化机制详解

作者：王廷云的博客

这篇文章主要介绍了RabbitMQ消息队列之持久化机制详解,持久化,即将原本存在于内存中的数据写入到磁盘上永久保存数据,防止服务宕机时内存数据的丢失,Rabbitmq 的持久化分为队列持久化、消息持久化和交换器持久化,需要的朋友可以参考下

一、RabbitMQ 持久化机制

1、RabbitMQ 持久化概述

持久化，即将原本存在于内存中的数据写入到磁盘上永久保存数据，防止服务宕机时内存数据的丢失。

Rabbitmq 的持久化分为队列持久化、消息持久化和交换器持久化。

对于消息来说，不管是持久化的消息还是非持久化的消息都可以被写入到磁盘。

持久化的消息会同时写入磁盘和内存（加快读取速度），非持久化消息会在内存不够用时，将消息写入磁盘（一般重启之后就没有了）。

2、队列持久化

队列的持久化是在定义队列时的通过 durable 参数来决定的，当 durable 为 true 时，才代表队列会持久化。例如：

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//第二个餐胡设置为true，代表队列持久化
channel.queueDeclare("queue.persistent.name", true, false, false, null);

关键的是第二个参数设置为 true，即 durable = true。Channel 类中 queueDeclare 的完整定义如下：

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
							Map<String, Object> arguments) throws IOException;

参数说明：

总结：如果将 queue 的持久化标识 durable 设置为 true，则代表是一个持久的队列。当服务重启之后，队列仍然会存在，这是因为服务会把持久化的 queue 存放在硬盘上，当服务重启的时候，会重新加载这些被持久化的 queue。

3、消息持久化

队列是可以被持久化，但是里面的消息是否为持久化那还要看消息的持久化设置。

也就是说，重启之前 queue 里面如果还有未发出去的消息的话，重启之后，消息是否还存在队列里面就要取决于在发送消息时对消息的设置。

消息持久化的实现需要在发送消息时设置消息的持久化标识，例如：

channel.basicPublish("exchange01", "routing_key01", MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent_message".getBytes());

方法原型是：

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;

这里关键的是 BasicProperties props 这个参数，它的定义如下：

public BasicProperties(
     String contentType,//消息类型如：text/plain
     String contentEncoding,//编码
     Map<String,Object> headers,
     Integer deliveryMode,//1:nonpersistent 2:persistent
     Integer priority,//优先级
     String correlationId,
     String replyTo,//反馈队列
     String expiration,//expiration到期时间
     String messageId,
     Date timestamp,
     String type,
     String userId,
     String appId,
     String clusterId
)

其中 deliveryMode=1 代表不持久化， deliveryMode=2 代表持久化。而代码实现中的 MessageProperties.PERSISTENT_PLAIN 值是官方提供的一个将 deliveryMode 设置为 2 的 BasicProperties 的对象：

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

除此之外，我们也可以使用另一种方式来设置消息持久化标志位：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.deliveryMode(2); // 将 deliveryMode 值设置为 2 表示消息持久化
AMQP.BasicProperties properties = builder.build();
channel.basicPublish("exchange01", "routing_key01", properties, "persistent_message".getBytes());

至此，我们可以知道：当 broker 服务其重启后，想要消息不丢失，既需要设置队列持久化，也需要设置消息持久化。

扩展知识：

basicPublish 方法还有另外两个重载方法：

void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body)
        throws IOException;
void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
        throws IOException;

这里有两个关键的参数： mandatory 和 immediate 。这两个标识位都有当消息传递过程中不可达目的地时将消息返回给生产者的功能。下面简单讲解下这两个标识位：

1）mandatory

2）immediate

概括来说就是：

4、交换器持久化

对于消息的可靠性来说，只需要设置队列的持久化和消息的持久化即可。exchange 的持久化并没有什么影响，但是，如果 exchange 不设置持久化的话，当 broker 服务重启之后，exchange 将不复存在，这样会导致消息发送者 producer 无法正常发送消息。

所以，建议同样设置 exchange 的持久化。而 exchange 的持久化设置也特别简单，设置方法原型如下：

Exchange.DeclareOk exchangeDeclare(String exchange,String type,boolean durable)throws IOException;

所以，只需要在声明的时候将 durable 字段设置为 true 即可：

channel.exchangeDeclare(exchangeName, “direct/topic/header/fanout”, true);

二、RabbitMQ 知识扩展

1、内存告警与内存换页

1-1、内存告警

当内存使用超过配置的阈值时，RabbitMQ 会暂时阻塞客户端的连接，并停止接收从客户端发来的消息，以避免服务崩溃，客户端与服务端的心跳检测也会失败。

当出现内存告警时，可以通过管理命令临时调整内存大小：

RabbitMQctl set_vm_memory_high_watermark <fraction>

通过此命令修改的阈值在 RabbitMQ 重启之后将会失效，如果想要设置的阈值永久有效需要修改配置文件：

# 相对值，也就是前面的fraction，建议设置在0.4~0.66之间，不要超过0.7
vm_memory_high_watermark.relative=0.4
# 绝对值，单位为KB,MB,GB,对应的临时命令是：RabbitMQctl set_vm_memory_high_watermark absolute <value>
#vm_memory_high_watermark.absolute=1GB

修改完配置文件后，需要重启服务才会生效。

1-2、内存换页

在某个 broker 节点触及内存阈值并阻塞生产者之前，它会尝试将队列内存中的消息换页存储到磁盘以释放内存空间。持久化和非持久化的消息都会被转储到磁盘中，其中持久化的消息本身就在磁盘中有一个备份，所以这里会将持久化的消息直接从内存中清除掉。

默认情况下，在内存使用达到设置的阈值的 50% 时会进行换页操作。也就是说，在默认的内存阈值 40% 的情况下，当内存超过 40% * 50% = 20% 时会经行换页动作。

内存换页阈值可以通过在配置文件中设置来进行调整：

vm_memory_high_watermark_paging_ratio=0.75

上面的配置将会在 RabbitMQ 内存使用率达到 30%（假设内存阈值是 0.4）时进行换页动作，并在 40% 时阻塞生产者（当 vm_memory_high_watermark_paging_ratio 的值大于 1 时，相当于禁用了换页功能）。

2、磁盘告警与配置

2-1、磁盘告警

当磁盘剩余空间低于设置的阈值时，RabbitMQ 同样会阻塞生产者，这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务崩溃。

默认情况下，磁盘的阈值是50M，表示当磁盘剩余空间低于50M时，会阻塞生产者并停止内存中消息的换页动作。

这个阈值的设置可以减小，但不能完全消除因磁盘耗尽而导致崩溃的可能性。比如在两次磁盘空间检测期间内，磁盘空间从大于50M被耗尽到0M。

2-2、修改磁盘告警阈值

可以通过以下命令临时调整磁盘阈值：

#设置具体大小，单位为KB/MB/GB
RabbitMQctl set_disk_free_limit <disk_limit>
#设置相对值，建议取值为1.0~2.0（相对于内存的倍数，如内存大小是8G,若为1.0，则表示磁盘剩余8G时，阻塞）
RabbitMQctl set_disk_free_limit mem_relative <fraction>

如果要永久生效需要对应的配置文件，配置如下（需要重启生效）：

disk_free_limit.relative=2.0
#disk_free_limit_absolute=50MB

这里有个建议：一个相对谨慎的做法是将磁盘阈值设置为与操作系统所显示的内存大小一致。

3、数据写入磁盘时机

4、磁盘消息格式

消息保存于 $MNESIA/msg_store_persistent/x.rdq 文件中，其中 x 为数字编号，从 1 开始，每个文件最大为 16M（16777216），超过这个大小会生成新的文件，文件编号加 1。

文件中的消息格式如下：

<<Size:64, MsgId:16/binary, MsgBody>>

5、磁盘文件删除机制

到此这篇关于RabbitMQ消息队列之持久化机制详解的文章就介绍到这了,更多相关RabbitMQ持久化机制内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文