亲宝软件园·资讯

展开

rabbitmq消息应答队列持久化消息持久化

专注写bug 人气:0

一、前言

Boolean autoAck = false;
channel.basicConsume(queue_name, autoAck ,consumer);

在simple queue 和 work queue(轮询) 处理中，我们设置的消费者的消息监听都采用 channel.basicConsume(queue_name,true, consumer)，其中参数二 boolean autoAck为true，但在fair prefetch 公平分发中设置为false，这个设置在整个消息队列和消息消费者之间有什么影响呢？

二、autoAck 参数的讨论

我们都明白一点，autoAck设置为true时，消息队列可以不用在意消息消费者是否处理完消息，一直发送全部消息。但在公平分发中，也就是autoAck设置为false，在发送一个消息后到没收到消息消费者成功消费消息的信息回执之间，是不会继续给这个消息继续发送消息的。

1、当 autoAck设置为true时，也就是自动确认模式，一旦消息队列将消息发送给消息消费者后，就会从内存中将这个消息删除。

2、当autoAck设置为false时，也就是手动模式，如果此时的有一个消费者宕机，消息队列就会将这条消息继续发送给其他的消费者，这样数据在消息消费者集群的环境下，也就算是不丢失了。

当设置autoAck为true时

在 Boolean autoAck = true的情况下，消息队列不会管消费者是否收到了消息，如果消费者宕机，消息也就丢失了。

当设置autoAck为false时

在 Boolean autoAck = false的情况下，如果消费者1宕机了，消息队列没有收到消费者发送回的应答，就会将这个消息发送给下一个消费者处理。直到消费者处理完这个消息，并向消息队列发送了一个消息应答，告诉消息队列此时这个消息已经处理完成，消息队列才会将这个消息从内存中删除。

由此我们可以思考一个问题：从上面两个设置中，我们当然会认为false比较好了，但大家可能会忽略一个小问题，消息队列保存的消息是在内存中的，消息队列宕机了，内存中的消息也就清除了，如何做到消息的持久化保存呢？

三、rabbitmq 队列持久化操作

我们之前在消息的生产者和消息的消费者中都声明了一个消息队列。

channel.queueDeclare(queue_name, false, false, false, null);

他的源码介绍为：

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

其中各项参数的含义：

queue：声明队列的名称
durable：如果我们声明一个持久队列，则为true（该队列将在服务器重启后保留下来）
exclusive：如果我们声明一个独占队列，则为true（仅限此连接）
autoDelete：如果我们声明一个自动删除队列（服务器将在不再使用它时将其删除）
arguments：队列的其他属性（构造参数）

【扩展：】arguments参数干啥用的？

当前的arguments参数的含义，用于设定队列的属性，如下所示：

1、x-expires设定队列有效期。表示队列在指定时间内未使用，则会删除。

2、x-message-ttl设定消息延迟发送时间。

3、x-dead-letter-exchange设置死信交换机。

4、x-dead-letter-routing-key设置路由。

参照文章：

RabbitMQ之消息有效期与死信

rabbitmq创建queue时arguments参数注释

小细节：从上面的参数信息中我们发现一个参数durable，发现这个参数是声明队列为持久化队列，那我们改成 true 是否就可以了呢？我们尝试下！

修改send代码中的声明队列：

boolean durable = true;
channel.queueDeclare(queue_name, durable, false, false, null);

运行起来，结果。。

在这里插入图片描述

注意：出现这种情况的原因是我的rabbitmq中本身就存在一个设置好了的queue，如下所示：

在这里插入图片描述

如果在已存在的消息队列上，依据修改代码变更持久化队列操作，则会出现如上所述的异常信息。
但如果rabbitmq中不存在对应的消息队列时，则不会造成影响。
结论：

rabbitmq不允许对一个已存在的队列重新定义参数信息。

由上面的测试发现：

1、如果在localhost：15672中删除指定的queue，则可以创建出一个持久化队列。
2、重新定义一个网址上不存在的名称作为持久化队列。

最后再强调一点：
消息生产者和消息消费者的队列声明(队列设置)，必须保持一致。
原因：rabbitmq不允许对一个已存在的队列重新定义参数信息
有些大佬说无需在消费者中声明队列，其实最好还是需要声明，原因在于，如果rabbitmq中不存在指定的queue_name的消息队列时，运行代码将会出现报错信息！！

四、2019.11.04 问题补充

上面的两个参数信息消息应答(autoAck)与消息持久化(durable)，都往持久化的方向设置了，消息会持久化保存吗？
答案：错。

1、消息应答设置为手动模式，只是确保消息能够正常的被消费掉，而并非标识消息的持久化。
2、durable设置为true，只是说我们设置一个消息队列的属性为持久化队列，在rabbitmq中有很多个通道和队列，并非标识整体的消息就是持久化了。

为什么说按照上述设置条件，设定了消息队列后，消息队列中的消息还是不能持久化保存呢？

消息生产者生产50个消息并放入消息队列中

在这里插入图片描述

重启rabbitmq服务(模拟宕机)

在这里插入图片描述

重启完成后，访问 localhost:15672 查看 Queues属性，

在这里插入图片描述

发现消息队列在重启服务后是存在的，但其中的消息却不存在了。

要想彻底实现服务宕机等操作后，消息依旧能够实现持久化保存(硬盘保存)，还需要继续进行学习，研究。

五、2019.11.07消息的持久化

通过上面的测试我们发现：

durable 只是表明消息队列的持久化，不表示消息的持久化。

在消息生产者生产消息推送至消息队列中(或消息转发器)时，我们使用了一个方法

channel.basicPublish(String exchange, 
String routingKey, 
BasicProperties props, 
byte[] body)

其中的参数三 BasicProperties props表示额外配置属性。
那么这个属性在源码中有什么呢？

public static class BasicProperties extends
com.rabbitmq.client.impl.AMQBasicProperties {
        private String contentType;//消息类型如：text/plain
        private String contentEncoding;//编码
        private Map<String,Object> headers;
        private Integer deliveryMode;//1:nonpersistent 2:persistent
        private Integer priority;//优先级
        private String correlationId;
        private String replyTo;//反馈队列
        private String expiration;//expiration到期时间
        private String messageId;
        private Date timestamp;
        private String type;
        private String userId;
        private String appId;
        private String clusterId;
...

从源码中我们看到 BasicProperties中存在一条属性 deliveryMode，

1表示不持久化； 2表示持久化！

如何使用代码实现呢？

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.BasicProperties.Builder;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import cn.linkpower.util.MqConnectUtil;

/**
 * 公平分发--谁做的快谁就多做！<br>
 * 只有在消息消费者成功消费消息，发送消费成功的指令给队列后，消息队列才会继续向该消费者发送下一条消息指令。<br>
 * @author 76519
 *
 */
public class Send {
	private static final String queue_name = "test_work_queue";
	public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
		//1、建立连接
		Connection mqConnection = MqConnectUtil.getMqConnection();
		//2、建立信道(通道)
		Channel channel = mqConnection.createChannel();
		//3、声明队列(开启持久化)
		boolean durable = true;
		channel.queueDeclare(queue_name, durable, false, false, null);
		
		//公平分发---
		//为了开启公平分发操作，在消息消费者发送确认收到的指示后，消息队列才会给这个消费者继续发送下一条消息。
		//此处的 1 表示 限制发送给每个消费者每次最大的消息数。
		channel.basicQos(1);
		
		//4、发送消息
		for (int i = 0; i < 10; i++) {
			String string = "hello xiangjiao "+i;
			System.out.println("send msg = "+string);
			//发送消息
			//channel.basicPublish("", queue_name, null, string.getBytes());
			//消息持久化测试
			Builder builder = new Builder();
			builder.deliveryMode(2);
			BasicProperties properties = builder.build();
			channel.basicPublish("", queue_name, properties, string.getBytes());
			//消息发送慢一点
			Thread.sleep(i*5);
		}
		//5、使用完毕后，需要及时的关闭流应用
		channel.close();
		mqConnection.close();
	}
}

测试操作：
1、运行代码，查看 local’host:15672 登陆指定的账号，查询queue信息

在这里插入图片描述

2、重启rabbitmq服务。

在这里插入图片描述

3、重启后，再次查看 web 控制台

在这里插入图片描述

发现：当重新完全启动 rabbitmq 后，他会自动加载之前的消息至消息队列中。

但是此时并不能说明问题，我们是否忽略了一点，你确定了这个消息队列的消息了没有？

so 我们运行消息消费者 查看这个消息队列里面的消息到底是什么？

在这里插入图片描述

在这里插入图片描述

六、2022.02.09 增加队列持久化说明

在之前的代码中，设置队列属性为createChannel.queueDeclare(simpleQueueName, false, false, false, null)，其中参数二代表该队列是否是一个持久化队列，此处设置的为false，表示非持久化。

这是什么含义呢?

执行消息创建并添加至队列的代码逻辑。此时队列中存在数据，其次也存在该队列。

在这里插入图片描述

将Rabbitmq重启，再次查看web：

/sbin/service rabbitmq-server stop
/sbin/service rabbitmq-server start

在这里插入图片描述

此时通过web界面得知，该队列在rabbitmq重启后，队列没了！

结语

到此为止，消息持久化、队列持久化 算是琢磨个差不多了。

加载全部内容
相关教程
猜你喜欢
用户评论