【实战】rabbitMQ Queue队列消息持久化[附源码]

小渣渣

rabbitMQ Queue队列消息默认不是持久化到硬盘的，意味着，一旦rabbitMQ服务重启，消息就会丢失，怎么处理呢？

队列的持久化

如过将queue的持久化标识durable设置为true,则代表是一个持久的队列，那么在服务重启之后，也会存在，因为服务会把持久化的queue存放在硬盘上，当服务重启的时候，会重新什么之前被持久化的queue。队列是可以被持久化，但是里面的消息是否为持久化那还要看消息的持久化设置。也就是说，重启之前那个queue里面还没有发出去的消息的话，重启之后那队列里面是不是还存在原来的消息，这个就要取决于发生着在发送消息时对消息的设置了。
如果要在重启后保持消息的持久化必须设置消息是持久化的标识。

设置队列持久化：

登录可见。

方法的第四的参数autoDelete，一般都会输入false。文档描述这个参数如果是true的话，意思是：如果这个queue不再使用（没有被订阅）的话，server就会删除它。在我的测试过程中，只要是连接改queue的所有接收者都断开连接的话，该queue就会被删除，即使里面还有没有处理的消息。RabbitMQ的重启也同样会删除他们。如果输入的是false，那与之相连的客户端都断开连接的话，服务是不会删除这个队列的，队列中的消息也就会存在。发送端在没有客户端连接的时候也可以把消息放入改队列，客户端起来的时候，就会得到这些消息。但是如果RabbitMQ服务重启的话，该队列就没有了，里面的消息自然也就没有了。

第三个参数是exclusive，文档描述说，如果是true，那么申明这个queue的connection断了，那么这个队列就被删除了，包括里面的消息。

第二个参数durable，文档描述说，如果是true，则代表是一个持久的队列，那么在服务重启后，也会存在。因为服务会把持久化的queue存放在硬盘上，放服务重启的时候，会重新申明这个queue。当然必须是在autoDelete和exclusive都为false的时候。队列是可以被持久化，但是里面的消息是否为持久化那还要看消息的持久化设置。也就是说，如果重启之前那个queue里面还有没有发出去的消息的话，重启之后那队列里面是不是还存在原来的消息，这个就要取决于发送者在发送消息时对消息的设置了。


我们修改完成代码后，尝试运行，会报错如下：

未经处理的异常:  RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text="PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'hello' in vhost 'myserver': received 'true' but current is 'false'", classId=50, methodId=10, cause=

因为我们已经定义了一个名叫hello的未持久化的队列。RabbitMQ不允许使用不同的参数设定重新定义已经存在的队列。

解决办法有两个：

1：重新声明一个不同名字的队列，比如my_queue
2：删除已定义的“hello”队列，地址为：http://localhost:15672，使用是用户名和密码登录。RabbitMQ的默认密码和用户名都是guest。点开“queue”那栏，可以看到队列列表，点击“hello”队列，会展开队列的详细信息。把页面拉到最后，有一项“Delete”，点开，点击“Delete Queue”按钮，就可以把队列删除掉了。 然后再运行代码的时候，就会创建一个支持持久化的hello队列。



消息的持久化

如果要在重启后保持消息的持久化必须设置这个消息是持久化的。设置是在发送者发送的时候，比较简单，代码如下：

登录可见。

DeliveryMode默认是1，不持久的，设置为2表示消息持久的

我们修改好代码以后，尝试只打开生产者程序发送消息，然后，重启rabbitMQ服务，再次打开消费者，发现消息并没有丢失。

（完）

附上c#源码：

游客，如果您要查看本帖隐藏内容请回复

mmxx0212

学习学习

15012764719

学习学习

531849028

怎么没看到Demo呢，一定要回复才能看到得到噻

gongjiacong

我要c#源码啊啊啊啊啊啊啊啊啊啊

yqxyrx

挺好 的，正是我所需要的

顾言

学习一下

坚持吧

1111111111111111

小渣渣

RabbitMQ的队列（Queue）的参数及其含义

/**
* Construct a new queue, given a name, durability flag, and auto-delete flag, and arguments.
* @param name the name of the queue - must not be null; set to "" to have the broker generate the name.
* @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 (the queue will only be used by the declarer's
* connection)
* @param autoDelete true if the server should delete the queue when it is no longer in use
* @param arguments the arguments used to declare the queue
*/
public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
   Assert.notNull(name, "'name' cannot be null");
   this.name = name;
   this.actualName = StringUtils.hasText(name) ? name
         : (Base64UrlNamingStrategy.DEFAULT.generateName() + "_awaiting_declaration");
   this.durable = durable;
   this.exclusive = exclusive;
   this.autoDelete = autoDelete;
   this.arguments = arguments != null ? arguments : new HashMap<>();
}

参数介绍：
1、name: 队列的名称；
2、actualName: 队列的真实名称，默认用name参数，如果name为空，则根据规则生成一个；
3、durable: 是否持久化；
4、exclusive: 是否独享、排外的；
5、autoDelete: 是否自动删除；
6、arguments：队列的其他属性参数，有如下可选项，可参看图2的arguments：
（1）x-message-ttl：消息的过期时间，单位：毫秒；
（2）x-expires：队列过期时间，队列在多长时间未被访问将被删除，单位：毫秒；
（3）x-max-length：队列最大长度，超过该最大值，则将从队列头部开始删除消息；
（4）x-max-length-bytes：队列消息内容占用最大空间，受限于内存大小，超过该阈值则从队列头部开始删除消息；
（5）x-overflow：设置队列溢出行为。这决定了当达到队列的最大长度时消息会发生什么。有效值是drop-head、reject-publish或reject-publish-dlx。仲裁队列类型仅支持drop-head；
（6）x-dead-letter-exchange：死信交换器名称，过期或被删除（因队列长度超长或因空间超出阈值）的消息可指定发送到该交换器中；
（7）x-dead-letter-routing-key：死信消息路由键，在消息发送到死信交换器时会使用该路由键，如果不设置，则使用消息的原来的路由键值
（8）x-single-active-consumer：表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，其他被忽略，false时消息循环分发给所有消费者(默认false)
（9）x-max-priority：队列要支持的最大优先级数;如果未设置，队列将不支持消息优先级；
（10）x-queue-mode（Lazy mode）：将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用;如果未设置，队列将保留内存缓存以尽可能快地传递消息；
（11）x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。