一、前言

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

在simple queue 和 work queue(輪詢) 處理中，我們設(shè)置的消費(fèi)者的消息監(jiān)聽都采用 channel.basicConsume(queue_name,true, consumer)，其中參數(shù)二 boolean autoAck為true，但在fair prefetch 公平分發(fā)中設(shè)置為false，這個設(shè)置在整個消息隊列和消息消費(fèi)者之間有什么影響呢？

二、autoAck 參數(shù)的討論

我們都明白一點(diǎn)，autoAck設(shè)置為true時，消息隊列可以不用在意消息消費(fèi)者是否處理完消息，一直發(fā)送全部消息。但在公平分發(fā)中，也就是autoAck設(shè)置為false，在發(fā)送一個消息后到?jīng)]收到消息消費(fèi)者成功消費(fèi)消息的信息回執(zhí)之間，是不會繼續(xù)給這個消息繼續(xù)發(fā)送消息的。

1、當(dāng) autoAck設(shè)置為true時，也就是自動確認(rèn)模式，一旦消息隊列將消息發(fā)送給消息消費(fèi)者后，就會從內(nèi)存中將這個消息刪除。

2、當(dāng)autoAck設(shè)置為false時，也就是手動模式，如果此時的有一個消費(fèi)者宕機(jī)，消息隊列就會將這條消息繼續(xù)發(fā)送給其他的消費(fèi)者，這樣數(shù)據(jù)在消息消費(fèi)者集群的環(huán)境下，也就算是不丟失了。

在 Boolean autoAck = true的情況下，消息隊列不會管消費(fèi)者是否收到了消息，如果消費(fèi)者宕機(jī)，消息也就丟失了。

在 Boolean autoAck = false的情況下，如果消費(fèi)者1宕機(jī)了，消息隊列沒有收到消費(fèi)者發(fā)送回的應(yīng)答，就會將這個消息發(fā)送給下一個消費(fèi)者處理。直到消費(fèi)者處理完這個消息，并向消息隊列發(fā)送了一個消息應(yīng)答，告訴消息隊列此時這個消息已經(jīng)處理完成，消息隊列才會將這個消息從內(nèi)存中刪除。

由此我們可以思考一個問題：從上面兩個設(shè)置中，我們當(dāng)然會認(rèn)為false比較好了，但大家可能會忽略一個小問題，消息隊列保存的消息是在內(nèi)存中的，消息隊列宕機(jī)了，內(nèi)存中的消息也就清除了，如何做到消息的持久化保存呢？

三、rabbitmq 隊列持久化操作

我們之前在消息的生產(chǎn)者和消息的消費(fèi)者中都聲明了一個消息隊列。

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;

其中各項參數(shù)的含義：

queue：聲明隊列的名稱
durable：如果我們聲明一個持久隊列，則為true（該隊列將在服務(wù)器重啟后保留下來）
exclusive：如果我們聲明一個獨(dú)占隊列，則為true（僅限此連接）
autoDelete：如果我們聲明一個自動刪除隊列（服務(wù)器將在不再使用它時將其刪除）
arguments：隊列的其他屬性（構(gòu)造參數(shù)）

【擴(kuò)展：】arguments參數(shù)干啥用的？

當(dāng)前的arguments參數(shù)的含義，用于設(shè)定隊列的屬性，如下所示：

1、x-expires設(shè)定隊列有效期。表示隊列在指定時間內(nèi)未使用，則會刪除。

2、x-message-ttl設(shè)定消息延遲發(fā)送時間。

3、x-dead-letter-exchange設(shè)置死信交換機(jī)。

4、x-dead-letter-routing-key設(shè)置路由。

參照文章：

RabbitMQ之消息有效期與死信

rabbitmq創(chuàng)建queue時arguments參數(shù)注釋

小細(xì)節(jié)：從上面的參數(shù)信息中我們發(fā)現(xiàn)一個參數(shù)durable，發(fā)現(xiàn)這個參數(shù)是聲明隊列為持久化隊列，那我們改成 true 是否就可以了呢？我們嘗試下！

修改send代碼中的聲明隊列：

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

運(yùn)行起來，結(jié)果。。

注意：出現(xiàn)這種情況的原因是我的rabbitmq中本身就存在一個設(shè)置好了的queue，如下所示：

如果在已存在的消息隊列上，依據(jù)修改代碼變更持久化隊列操作，則會出現(xiàn)如上所述的異常信息。
但如果rabbitmq中不存在對應(yīng)的消息隊列時，則不會造成影響。
結(jié)論：

rabbitmq不允許對一個已存在的隊列重新定義參數(shù)信息。

由上面的測試發(fā)現(xiàn)：

1、如果在localhost：15672中刪除指定的queue，則可以創(chuàng)建出一個持久化隊列。
2、重新定義一個網(wǎng)址上不存在的名稱作為持久化隊列。

最后再強(qiáng)調(diào)一點(diǎn)：
消息生產(chǎn)者和消息消費(fèi)者的隊列聲明(隊列設(shè)置)，必須保持一致。
原因：rabbitmq不允許對一個已存在的隊列重新定義參數(shù)信息
有些大佬說無需在消費(fèi)者中聲明隊列，其實最好還是需要聲明，原因在于，如果rabbitmq中不存在指定的queue_name的消息隊列時，運(yùn)行代碼將會出現(xiàn)報錯信息??！

四、2019.11.04 問題補(bǔ)充

上面的兩個參數(shù)信息消息應(yīng)答(autoAck)與消息持久化(durable)，都往持久化的方向設(shè)置了，消息會持久化保存嗎？
答案：錯。

1、消息應(yīng)答設(shè)置為手動模式，只是確保消息能夠正常的被消費(fèi)掉，而并非標(biāo)識消息的持久化。
2、durable設(shè)置為true，只是說我們設(shè)置一個消息隊列的屬性為持久化隊列，在rabbitmq中有很多個通道和隊列，并非標(biāo)識整體的消息就是持久化了。

為什么說按照上述設(shè)置條件，設(shè)定了消息隊列后，消息隊列中的消息還是不能持久化保存呢？

消息生產(chǎn)者生產(chǎn)50個消息并放入消息隊列中

重啟rabbitmq服務(wù)(模擬宕機(jī))

重啟完成后，訪問 localhost:15672 查看 Queues屬性，

發(fā)現(xiàn)消息隊列在重啟服務(wù)后是存在的，但其中的消息卻不存在了。

要想徹底實現(xiàn)服務(wù)宕機(jī)等操作后，消息依舊能夠?qū)崿F(xiàn)持久化保存(硬盤保存)，還需要繼續(xù)進(jìn)行學(xué)習(xí)，研究。

五、2019.11.07消息的持久化

通過上面的測試我們發(fā)現(xiàn)：

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

在消息生產(chǎn)者生產(chǎn)消息推送至消息隊列中(或消息轉(zhuǎn)發(fā)器)時，我們使用了一個方法

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

其中的參數(shù)三 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;//優(yōu)先級
        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表示持久化！

如何使用代碼實現(xiàn)呢？

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;

/**
 * 公平分發(fā)--誰做的快誰就多做！<br>
 * 只有在消息消費(fèi)者成功消費(fèi)消息，發(fā)送消費(fèi)成功的指令給隊列后，消息隊列才會繼續(xù)向該消費(fèi)者發(fā)送下一條消息指令。<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);
		
		//公平分發(fā)---
		//為了開啟公平分發(fā)操作，在消息消費(fèi)者發(fā)送確認(rèn)收到的指示后，消息隊列才會給這個消費(fèi)者繼續(xù)發(fā)送下一條消息。
		//此處的 1 表示 限制發(fā)送給每個消費(fèi)者每次最大的消息數(shù)。
		channel.basicQos(1);
		
		//4、發(fā)送消息
		for (int i = 0; i < 10; i++) {
			String string = "hello xiangjiao "+i;
			System.out.println("send msg = "+string);
			//發(fā)送消息
			//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());
			//消息發(fā)送慢一點(diǎn)
			Thread.sleep(i*5);
		}
		//5、使用完畢后，需要及時的關(guān)閉流應(yīng)用
		channel.close();
		mqConnection.close();
	}
}

測試操作：
1、運(yùn)行代碼，查看 local’host:15672 登陸指定的賬號，查詢queue信息

2、重啟rabbitmq服務(wù)。

3、重啟后，再次查看 web 控制臺

發(fā)現(xiàn)：當(dāng)重新完全啟動 rabbitmq 后，他會自動加載之前的消息至消息隊列中。

但是此時并不能說明問題，我們是否忽略了一點(diǎn)，你確定了這個消息隊列的消息了沒有？

so 我們運(yùn)行消息消費(fèi)者 查看這個消息隊列里面的消息到底是什么？

六、2022.02.09 增加隊列持久化說明

在之前的代碼中，設(shè)置隊列屬性為createChannel.queueDeclare(simpleQueueName, false, false, false, null)，其中參數(shù)二代表該隊列是否是一個持久化隊列，此處設(shè)置的為false，表示非持久化。

這是什么含義呢?

執(zhí)行消息創(chuàng)建并添加至隊列的代碼邏輯。此時隊列中存在數(shù)據(jù)，其次也存在該隊列。

將Rabbitmq重啟，再次查看web：

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

此時通過web界面得知，該隊列在rabbitmq重啟后，隊列沒了！

結(jié)語

到此為止，消息持久化、隊列持久化 算是琢磨個差不多了。

到此這篇關(guān)于rabbitmq學(xué)習(xí)系列教程之消息應(yīng)答(autoAck)、隊列持久化(durable)及消息持久化的文章就介紹到這了,更多相關(guān)rabbitmq消息應(yīng)答隊列持久化消息持久化內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！