RabbitMQ的消息持久化處理
- 2019 年 11 月 5 日
- 筆記
1、RabbitMQ的消息持久化處理，消息的可靠性是 RabbitMQ 的一大特色，那麼 RabbitMQ 是如何保證消息可靠性的呢——消息持久化。
2、autoDelete屬性的理解。
1）、@Queue: 當autoDelete屬性設置到該註解的時候，含義即是，當所有消費者客戶端連接斷開後，是否自動刪除隊列，當設置值是true的時候刪除該隊列，當值是false的時候不刪除該隊列。
2）、@Exchange：當autoDelete屬性設置到該註解的時候，含義即是，當所有綁定隊列都不在使用時，是否自動刪除交換器，當設置值是true的時候刪除該交換器，當值是false的時候不刪除該交換器。
3、之前寫過RabbitMQ的交換器Exchange之direct（發佈與訂閱 完全匹配），這裡藉助這個進行消息持久化測試。生產者的代碼不發生改變，這裡只是將消費者的autoDelete = "true"屬性修改為autoDelete = "false"，進行對比測試。
Info級別的日誌進行消息的持久化操作，即隊列不進行自動刪除。將autoDelete = "false"即可。
1 package com.example.bie.consumer; 2 3 import org.springframework.amqp.core.ExchangeTypes; 4 import org.springframework.amqp.rabbit.annotation.Exchange; 5 import org.springframework.amqp.rabbit.annotation.Queue; 6 import org.springframework.amqp.rabbit.annotation.QueueBinding; 7 import org.springframework.amqp.rabbit.annotation.RabbitHandler; 8 import org.springframework.amqp.rabbit.annotation.RabbitListener; 9 import org.springframework.stereotype.Component; 10 11 /** 12 * 13 * @author biehl 14 * 15 * 消息接收者 16 * 17 * 1、@RabbitListener bindings:綁定隊列 18 * 19 * 2、@QueueBinding 20 * value:綁定隊列的名稱、exchange:配置交換器、key:路由鍵routing-key綁定隊列和交換器 21 * 22 * 3、@Queue value:配置隊列名稱、autoDelete:是否是一個可刪除的臨時隊列 23 * 24 * 4、@Exchange value:為交換器起個名稱、type:指定具體的交換器類型 25 * 26 * 27 */ 28 @Component 29 @RabbitListener(bindings = @QueueBinding( 30 31 value = @Queue(value = "${rabbitmq.config.queue.info}", autoDelete = "false"), 32 33 exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.DIRECT), 34 35 key = "${rabbitmq.config.queue.info.routing.key}")) 36 public class LogInfoConsumer { 37 38 /** 39 * 接收消息的方法,採用消息隊列監聽機制. 40 * 41 * @RabbitHandler意思是將註解@RabbitListener配置到類上面 42 * 43 * @RabbitHandler是指定這個方法可以進行消息的接收並且消費. 44 * 45 * @param msg 46 */ 47 @RabbitHandler 48 public void consumer(String msg) { 49 // 打印消息 50 System.out.println("INFO消費者===>消費: " + msg); 51 } 52 53 }
Error級別的日誌進行消息的持久化操作，即隊列進行自動刪除。將autoDelete = "true"即可。
1 package com.example.bie.consumer; 2 3 import org.springframework.amqp.core.ExchangeTypes; 4 import org.springframework.amqp.rabbit.annotation.Exchange; 5 import org.springframework.amqp.rabbit.annotation.Queue; 6 import org.springframework.amqp.rabbit.annotation.QueueBinding; 7 import org.springframework.amqp.rabbit.annotation.RabbitHandler; 8 import org.springframework.amqp.rabbit.annotation.RabbitListener; 9 import org.springframework.stereotype.Component; 10 11 /** 12 * 13 * @author biehl 14 * 15 * 消息接收者 16 * 17 * 1、@RabbitListener bindings:綁定隊列 18 * 19 * 2、@QueueBinding 20 * value:綁定隊列的名稱、exchange:配置交換器、key:路由鍵routing-key綁定隊列和交換器 21 * 22 * 3、@Queue value:配置隊列名稱、autoDelete:是否是一個可刪除的臨時隊列 23 * 24 * 4、@Exchange value:為交換器起個名稱、type:指定具體的交換器類型 25 * 26 * 27 */ 28 @Component 29 @RabbitListener(bindings = @QueueBinding( 30 31 value = @Queue(value = "${rabbitmq.config.queue.error}", autoDelete = "true"), 32 33 exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.DIRECT), 34 35 key = "${rabbitmq.config.queue.error.routing.key}")) 36 public class LogErrorConsumer { 37 38 /** 39 * 接收消息的方法,採用消息隊列監聽機制. 40 * 41 * @RabbitHandler意思是將註解@RabbitListener配置到類上面 42 * 43 * @RabbitHandler是指定這個方法可以進行消息的接收並且消費. 44 * 45 * @param msg 46 */ 47 @RabbitHandler 48 public void consumer(String msg) { 49 // 打印消息 50 System.out.println("ERROR消費者===>消費<===消息message: " + msg); 51 } 52 53 }
4、啟動你的生產者，啟動你的消費者，觀察RabbitMQ的圖形化界面。未生產消息、未消費消息的界面如下所示：
生產消息、消費消息的界面如下所示，我這裡還使用瀏覽器訪問控制層觸發生產者生產消息，消費者消費消息：
現在停止你的消費者，記錄消息到第幾條消息了。方便再次啟動消費者進行觀察。
啟動你的消費者，觀察，看看是從第幾條開始消費的。可以看到消息從第82條開始消費的。
RabbitMQ的消息持久化處理，Ready是對未接收到的數據狀態表示，如果RabbitMQ在隊列裏面存放的消息未被消費者所消費，那麼會給未消費的消息加一個標記，表示當前這個消息未被消費。消息持久化處理解決了丟失消息的這種狀況，我們可以接收到消息，就是因為隊列一直存在着呢，但是手動刪除隊列，消息也就丟失了，所以要慎重操作。當消費者停止以後，生產者生產的消息存儲在RabbitMQ的服務器內存中，隊列也存在內存中，數據在隊列中，即數據保存在內存中。但是如果RabbitMQ的服務都停止了，隊列也就消失了，隊列消失了，數據也就丟失了。
