轉自： https://www.jianshu.com/p/94d6d5d98c3d

?

序言

現在我們每天都要與信息打交道，主動或被動的在創造或接收消息。你會收到話費通知短信，使用微信 QQ跟遠在萬里的朋友交流，也可能使用釘釘跟同事討論工作，使用抖音娛樂等等。信息要準確及時的發送和接收 這背后使用了 消息隊列的相關技術。本文以RabbitMQ為例 講解消息隊列涉及的相關技術及使用場景，結合自身開發經驗幫助讀者更好理解這個隱藏在背后的這項“黑科技”

概括

常見的消息通訊方式有同步和異步兩種，消息隊列實現了異步通訊方式，即消息先存儲到消息隊列容器里 ，滿足某種條件后 發送給消息的接收方，當然消息隊列也可以實現同步通訊。隊列(Queue)是一種數據結構 滿足了先進先出的規則。這個消息的傳輸有生產者 交換機 消息隊列 消費者4部分組成。

RabbitMQ是一種易擴展，高可用 實現了AMQP協議并被廣泛應用的開源消息隊列。

?

image

RabbitMQ隊列屬性簡介

AMQP的屬性在RabbitMQ體現:

1 持久化 即隊列持久化保存到硬盤，如果服務重啟 可以再次恢復隊列

如在代碼里配置兩個 Queue

/*** queue 持久化* @return*/ @Bean public Queue DurableTrueQueue(){return new Queue("durableQueue",true); }/*** queue 非持久化* @return*/ @Bean public Queue DurableFalseQueue(){return new Queue("noDurableQueue",false); }

如注釋那樣 隊列durableQueue 是持久化隊列 而 noDurableQueue是非持久化隊列

在RabbmitMQ的管理后臺可以看到這兩個隊列信息

?

image

重啟RabbitMQ服務器 可以看到noDurableQueue 已經沒了

?

image

2 獨有性 排他性 即隊列只為當前鏈接服務 鏈接斷開 隊列被刪除

我對獨有性理解就是 隊列只服務于一個鏈接 鏈接消失 則隊列也被刪除

代碼里配置排他性的 隊列

?

@Bean public Queue exclusiveTrueQueue(){return new Queue("exclusiveTrueQueue",true,true,false); }

在RabbitMQ后臺管理頁面可以看到 該隊列

?

image

點擊隊列名稱 exclusiveTrueQueue 進入隊列詳情頁面

點擊get Messages 里的 get Message 按鈕 會報錯 說該隊列具有排他性 不能獲取消息內容

?

image

3 自動刪除 即消費者斷開 隊列被刪除

自動刪除 是指如果該隊列對應的鏈接全部斷開 則刪除該隊列 讀者可參考自行實驗

需要說明的是 如果隊列的 exclusive 為true或auto delete為true 那durable屬性是不起作用的 因為服務器重啟 鏈接都會斷掉 隊列信息會被刪除

4 其他可選隊列參數參數屬性(如隊列長度 過期時間等)

仲裁隊列

在集群環境下，仲裁隊列可以提供一種高效的 保證數據安全的 數據傳輸能力，能保證在某一臺服務器不可用情況下 主從服務的快速切換 和數據完整性復制 從而達到RabbitMQ的高可用和高性能,仲裁隊列遵循Raft分布式協議。仲裁隊列是RabbitMQ3.8.*版本以后加入的隊列類型 用以替換之前版本的隊列鏡像

在RabbitMQ集群架構中 ，會有一個主實例和多個從實例，主實例負責跟發送端 接收端的交互，把接收到的數據 往從實例復制一份，即從實例是主實例的拷貝和備份，當主實例節點宕機或不可用時候 從實例中會選出一個新的主實例 進行消息的接收和發送 從而保證隊列的可用性。

為保證主從節點數據一致性 ，只有當主節點的數據全部寫入從節點時候，主節點才會跟發送方確認消息已接收。數據完整保存或最接近完整數據的從節點才有可能被選舉為主節點 ，數據殘缺不全或者數據要比其他節點少的從節點是不會被選為主節點的，即從節點的選擇要最大可能保證數據的完整性。

仲裁隊列創建:

?

@Bean Queue quorumQueue(){Map<String,Object> map = new HashMap<>();map.put("x-queue-type","quorum");return new Queue("quorumQueue",true,false,false,map); }

在RabbitMQ管理后臺可以看到已經創建成功：

?

image

消息過期時間(Time-To-Live)

RabbitMQ 隊列里的消息如果超過了過期時間沒有消費者接收就變成了死信。處于消息隊列中的死信不能發送給客戶端，消息服務器會在過期后將消息刪除。

RabbitMQ 的過期時間(以下簡稱TTL)有兩種類型設置:

1 設置單條消息的過期時間

2 以隊列為單位設置消息的過期時間(代碼，命令行兩種方式)

兩種設置時間的單位都是毫秒

以隊列為單位設置消息的過期時間 需要在聲明隊列時候 設置消息過期參數

?

@Bean public Queue ttlQueue(){Map<String,Object> map = new HashMap<>();map.put("x-message-ttl",60000);return new Queue("ttlQueue",true,false,false,map); }

如果x-message-ttl 設置為0，則該隊列的消息需要同步接收 不能在隊列里保存 否則會過期

設置成功后 在服務器管理頁面可以看到TTL標識

?

image

設置單條消息過期時間

?

image

需要注意的是 MessageProperties 的Expiration類型為String格式

隊列過期時間

隊列過期時間指超過一定時間 隊列沒有被消費者消費 也沒有被重新聲明續租 消息隊列節點會自動刪除該隊列

隊列過期可以使用在 不斷創建新隊列 老的隊列在沒有一定時間沒有使用后會自動刪除釋放資源，如 聊天場景消息的發送接收 RPC 通過不斷創建隊列傳輸消息 當聊天結束后 自動刪除隊列

如果對持久化隊列設置了過期時間 當服務器重啟后 過期時間會重新開始計算

設置方式是 在創建聲明隊列時候 傳入參數x-expires

?

@Bean public Queue ttlForQueue(){Map<String,Object> map = new HashMap<>();map.put("x-expires",60000);return new Queue("ttlForQueue",true,false,false,map); }

?

image

隊列長度限制

隊列的最大長度既可以限制隊列里處理消息的數量 也可以限制處理消息總的字節大小。

這里的長度限制是指被消費者處理過的消息累積，不包括沒有被消費者端接收的在途消息

當消息達到隊列長度限制的時候，系統默認會將隊列頭部(最老的消息)刪除或設置為死信。

可以設置消息超限的應對策略 參數為x-overflow 值對應為

1 drop-head(默認方式 刪除隊列最早消息)

2 reject-publish(拒絕新的消息加入)

@RequestMapping(value = "/limitQueue",method = RequestMethod.GET) public void sendMessageForLimitQueue(){Map<String,Object> args = new HashMap<>();args.put("x-overflow","drop-head");args.put("x-max-length",5);//聲明數量受限的隊列Queue queue = new Queue("limitQueue",true,false,false,args);rabbitAdmin.declareQueue(queue);//綁定到交換機Binding binding = new Binding("limitQueue",Binding.DestinationType.QUEUE,"testExchange","limitQueueRoutkey",null);rabbitAdmin.declareBinding(binding);String content = "hello this is a test message ";Map<String,Object> map = makeMessage();map.put("content",content);System.out.println("開始發送。。"+map.toString());rabbitTemplate.convertAndSend("testExchange","limitQueueRoutkey",map); }

?

?

?

總結

以上是生活随笔為你收集整理的转：RabbitMQ 消息队列特性知多少的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。