腳本之家服務器常用軟件

快捷導航

rabbitmq學習系列教程之消息應答(autoAck)、隊列持久化(durable)及消息持久化

更新時間：2022年03月01日 11:02:23 作者：專注寫bug

這篇文章主要介紹了rabbitmq學習系列教程之消息應答(autoAck)、隊列持久化(durable)及消息持久化,本文通過示例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下

一、前言

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

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

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

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

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

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

當設置autoAck為true時

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

當設置autoAck為false時

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

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

三、rabbitmq 隊列持久化操作

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

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（該隊列將在服務器重啟后保留下來）
exclusive：如果我們聲明一個獨占隊列，則為true（僅限此連接）
autoDelete：如果我們聲明一個自動刪除隊列（服務器將在不再使用它時將其刪除）
arguments：隊列的其他屬性（構造參數(shù)）

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

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

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

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

3、x-dead-letter-exchange設置死信交換機。

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

參照文章：

RabbitMQ之消息有效期與死信

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

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

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

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

運行起來，結果。。

在這里插入圖片描述

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

在這里插入圖片描述

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

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

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

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

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

四、2019.11.04 問題補充

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

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

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

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

在這里插入圖片描述

重啟rabbitmq服務(模擬宕機)

在這里插入圖片描述

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

在這里插入圖片描述

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

要想徹底實現(xiàn)服務宕機等操作后，消息依舊能夠實現(xiàn)持久化保存(硬盤保存)，還需要繼續(xù)進行學習，研究。

五、2019.11.07消息的持久化

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

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

在消息生產(chǎn)者生產(chǎ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ā)送消費成功的指令給隊列后，消息隊列才會繼續(xù)向該消費者發(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ā)送確認收到的指示后，消息隊列才會給這個消費者繼續(xù)發(fā)送下一條消息。
		//此處的 1 表示 限制發(fā)送給每個消費者每次最大的消息數(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ā)送慢一點
			Thread.sleep(i*5);
		}
		//5、使用完畢后，需要及時的關閉流應用
		channel.close();
		mqConnection.close();
	}
}

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

在這里插入圖片描述