前言

  本篇是Rabbit MQ高級特性的學(xué)習(xí)筆記，記錄RabbitMQ的Classic，Quorum，Stream隊(duì)列，懶隊(duì)列的特性和運(yùn)用場景

一、Rabbit MQ隊(duì)列的選擇

  Rabbit MQ默認(rèn)提供了三種隊(duì)列：

• Classic：經(jīng)典隊(duì)列，在單機(jī)模式下是最常用的。
• Quorum：仲裁隊(duì)列，通常用于集群環(huán)境下，保證集群的高可用性。
• Stream：流式隊(duì)列，是自3.9.0版本開始引入的新特性，這種隊(duì)列類型的消息是持久化到磁盤并且具備分布式備份的。

1.1、Classic

經(jīng)典隊(duì)列是Rabbit MQ默認(rèn)的隊(duì)列，與Spring Boot整合時，使用new Queue，創(chuàng)建的隊(duì)列就是普通隊(duì)列：

經(jīng)典隊(duì)列在創(chuàng)建時除了指定隊(duì)列的名稱，還有額外的四個選項(xiàng)：

對應(yīng)的是頁面上的：

durable代表了隊(duì)列是否進(jìn)行持久化，如果開啟持久化，則會將隊(duì)列保存到磁盤上，將來Rabbit MQ重啟后，可以從磁盤恢復(fù)，否則隊(duì)列只存在于內(nèi)存中，服務(wù)重啟后會自動刪除。這里的durable只是設(shè)置了隊(duì)列的持久化。消息和交換機(jī)，同樣可以設(shè)置持久化。如果消息設(shè)置了持久化，而隊(duì)列未設(shè)置持久化，重啟之后，消息依舊會丟失。所以如果要保證消息的零丟失，消息和隊(duì)列都需要設(shè)置持久化。

// 隊(duì)列聲明
channel.queueDeclare("safe_queue", true, false, false, null); // Durability=true
// 消息發(fā)布
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // 持久化消息
    .build();

  exclusive代表了隊(duì)列是否排他。當(dāng)屬性為true時，表示僅允許聲明了該隊(duì)列的連接進(jìn)行訪問，其他連接無法訪問該隊(duì)列。通常用于單消費(fèi)者模式，確保隊(duì)列只能被特定消費(fèi)者訪問。
  autoDelete代表了隊(duì)列是否自動刪除，如果設(shè)置為true，表示該隊(duì)列在最后一個消費(fèi)者斷開連接之后，進(jìn)行刪除操作。通常用于臨時隊(duì)列的場景。
  arguments可以指定更多的參數(shù)，比如指定死信交換機(jī)和路由鍵，消息過期時間，最大長度，是否為懶隊(duì)列等。

1.2、Quorum

Quorum是針對鏡像隊(duì)列的一種優(yōu)化，目前已經(jīng)取代了鏡像隊(duì)列，作為Rabbit MQ集群部署保證高可用性的解決方案。傳統(tǒng)的鏡像隊(duì)列，是將消息副本存儲在一組節(jié)點(diǎn)上，以提高可用性和可靠性。鏡像隊(duì)列將隊(duì)列中的消息復(fù)制到一個或多個其他節(jié)點(diǎn)上，并使這些節(jié)點(diǎn)上的隊(duì)列保持同步。當(dāng)一個節(jié)點(diǎn)失敗時，其他節(jié)點(diǎn)上的隊(duì)列不受影響，因?yàn)樗鼈兩厦娑加邢⒌膫浞荨?br />  鏡像隊(duì)列使用主從模式，所有消息寫入和讀取均通過主節(jié)點(diǎn)，并異步復(fù)制到鏡像節(jié)點(diǎn)。主節(jié)點(diǎn)故障時需重新選舉，期間隊(duì)列不可用。而仲裁隊(duì)列基于Raft分布式共識算法，所有節(jié)點(diǎn)組成仲裁組。消息需被多數(shù)節(jié)點(diǎn)持久化后才確認(rèn)成功，Leader故障時自動觸發(fā)選舉。
  相比較于傳統(tǒng)的主從模式，避免了發(fā)生網(wǎng)絡(luò)分區(qū)時的腦裂問題（基于Raft分布式共識算法避免）。

和普通隊(duì)列的區(qū)別：

相比較于普通隊(duì)列，仲裁隊(duì)列增加了一個對于有毒消息的處理。什么是有毒消息？首先，消費(fèi)者從隊(duì)列中獲取到了元素，隊(duì)列會將該元素刪除，但是消費(fèi)者消費(fèi)失敗了，會給隊(duì)列nack，并且可以設(shè)置消息重新入隊(duì)。這樣可能存在因?yàn)闃I(yè)務(wù)代碼的問題，某條消息一直處理不成功的問題。仲裁隊(duì)列會記錄消息的重新投遞次數(shù)，判斷是否超過了設(shè)置的閾值，如果超過了就直接丟棄，或者放入死信隊(duì)列人工處理。
如果需要聲明一個仲裁隊(duì)列，只需要加入?yún)?shù)：

@Configuration
public class QuorumConfig {
    @Bean
    public Queue quorumQueue() {
        Map<String,Object> params = new HashMap<>();
        params.put("x-queue-type","quorum");
        return new Queue(MyConstants.QUEUE_QUORUM,true,false,false,params);
    }
}

  仲裁隊(duì)列適用于集群環(huán)境下，隊(duì)列長期存在，并且對于消息可靠性要求高，允許犧牲一部分性能（因?yàn)閞aft算法，消息需被多數(shù)節(jié)點(diǎn)持久化后才確認(rèn)成功）的場景。

1.3、Stream

在傳統(tǒng)的隊(duì)列模型中，同一條消息只能被一個消費(fèi)者消費(fèi)（一個隊(duì)列如果有多個消費(fèi)者，是工作分發(fā)的機(jī)制。消息1->消費(fèi)者1，消息2->消費(fèi)者2，消息3->消費(fèi)者1，不能兩個消費(fèi)者讀同一條消息。），并且消息是閱后即焚的（消費(fèi)者接收到消息后，隊(duì)列中的該消息就刪除，如果消費(fèi)者拒絕簽收并且設(shè)置了重新入隊(duì)，再把消息重新放入隊(duì)列中），無法重復(fù)從隊(duì)列中獲取相同的消息。并且在當(dāng)隊(duì)列中積累的消息過多時，性能下降會非常明顯。
Stream隊(duì)列正是解決了以上的這些問題。Stream隊(duì)列的核心是用aof文件的形式存儲隊(duì)列，將消息以aof的方式追加到文件中。允許用戶在日志的任何一個連接點(diǎn)開始重新讀取數(shù)據(jù)。（需要用戶自己記錄偏移量）
聲明一個stream隊(duì)列：

@Configuration
public class StreamConfig {
    @Bean
    public Queue streamQueue() {
        Map<String,Object> params = new HashMap<>();
        params.put("x-queue-type","stream");
        params.put("x-max-length-bytes", 20_000_000_000L); // 指定隊(duì)列的大小
        params.put("x-stream-max-segment-size-bytes", 100_000_000); // 文件分片存儲，每一片的大小
			 //必須設(shè)置持久化為true，同時獨(dú)占和自動刪除模式為false
        return new Queue(MyConstants.QUEUE_STREAM,true,false,false,params);
    }
}

  聲明消費(fèi)者：

public void stremReceiver(Channel channel,String message){
		try {
			channel.basicQos(100);
			Consumer myconsumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope,
										   AMQP.BasicProperties properties, byte[] body)
						throws IOException {
					System.out.println("========================");
					String routingKey = envelope.getRoutingKey();
					System.out.println("routingKey >"+routingKey);
					String contentType = properties.getContentType();
					System.out.println("contentType >"+contentType);
					long deliveryTag = envelope.getDeliveryTag();
					System.out.println("deliveryTag >"+deliveryTag);
					System.out.println("content:"+new String(body,"UTF-8"));
					// (process the message components here ...)
					//消息處理完后，進(jìn)行答復(fù)。答復(fù)過的消息，服務(wù)器就不會再次轉(zhuǎn)發(fā)。
					//沒有答復(fù)過的消息，服務(wù)器會一直不停轉(zhuǎn)發(fā)。
					channel.basicAck(deliveryTag, false);
				}
			};
			Map<String,Object> consumeParam = new HashMap<>();
			//first: 從日志隊(duì)列中第一個可消費(fèi)的消息開始消費(fèi)
			//last: 消費(fèi)消息日志中最后一個消息
			//next: 相當(dāng)于不指定offset，消費(fèi)不到消息。
			//Offset: 一個數(shù)字型的偏
			//Timestamp:一個代表時間的Data類型變量，表示從這個時間點(diǎn)開始消費(fèi)。
			//例如 一個小時前 Date timestamp = new Date(System.currentTimeMillis() - 60 * 60 * 1_000)
			consumeParam.put("x-stream-offset","last");
			channel.basicConsume(MyConstants.QUEUE_STREAM, false,consumeParam, myconsumer);
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("quorumReceiver received message : "+ message);
	}

1.4、懶隊(duì)列

Rabbit MQ對于常規(guī)隊(duì)列的處理是，將消息優(yōu)先存在于內(nèi)存中，在合適的時機(jī)再持久化到磁盤上，而懶隊(duì)列則相反，懶隊(duì)列會盡可能早的將消息內(nèi)容保存到磁盤當(dāng)中，并且只有在用戶請求到時，才臨時從磁盤加載到內(nèi)存當(dāng)中。懶隊(duì)列的設(shè)計(jì)也是為了應(yīng)對消息堆積問題的。
聲明懶隊(duì)列的方式，只需要加入?yún)?shù)，相應(yīng)的，當(dāng)一個隊(duì)列被聲明為懶隊(duì)列，那即使隊(duì)列被設(shè)定為不持久化，消息依然會寫入到硬盤中。

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-queue-mode", "lazy");

懶隊(duì)列適合消息量大且長期有堆積的隊(duì)列，可以減少內(nèi)存使用，加快消費(fèi)速度。

到此這篇關(guān)于RabbitMQ隊(duì)列的選擇的文章就介紹到這了,更多相關(guān)RabbitMQ隊(duì)列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論