四種途徑提高RabbitMQ傳輸數據的可靠性

前言

RabbitMQ雖然有對隊列及消息等的一些持久化設置，但其實光光只是這一個是不可以保障數據的可靠性的，下面咱們提出這樣的質疑：

（1）RabbitMQ生產者是不知道本身發佈的消息是否已經正確達到服務器呢，若是中間發生網絡異常等狀況呢？消息必然會丟失！

（2）RabbitMQ若是沒有設置隊列持久化，RabbitMQ服務器重後隊列的元數據會丟失，消息天然也會丟失！

（3）RabbitMQ若是消費者設置自動確認，即autoAck爲true，那麼無論消費者發生什麼狀況，該消息會自動從隊列中移除，實際上消費者有可能掛掉，消息必然會丟失！

（4）RabbitMQ中的消息若是沒有匹配到隊列時，那麼消息也會丟失！

本文其實也就是結合以上四個方面進行講解的，主要參考《RabbitMQ實戰指南》（有須要PDF電子書的能夠評論或者私信我），本文截圖也來自其中，另外能夠對一些RabbitMQ的概念的認識能夠參考個人上兩篇博文認識RabbitMQ交換機模型、RabbitMQ是如何運轉的？

 

---

 

 

1、設置mandotory參數、AE備份交換器

針對前言中的第（4）個問題，咱們能夠經過設置mandotory參數與AE備份交換器來解決

一、mandotory參數

　　1）當爲true時，交換器沒法根據自身的類型和路由鍵找到一個符合條件的隊列，此時RabbitMQ會調用Basic.Return命令將消息返回給生產者，消息將不會丟失

　　2）當爲false時，消息將會被直接丟棄。

　　3）RabbitMQ經過addReturnListener添加ReturnLisener監聽器監聽獲取沒有被正確路由到合適隊列的消息。

channel.basicPublish(EXCHANGE NAME, "", true, 
MessageProperties.PERSISTENT_TEXT_PLAIN, 
"mandatory test".getBytes()); 
channel.addReturnListener(new ReturnListener(){ 
　　public void handleReturn(int replyCode, String replyText, 
　　　　　　　　　　　　　　　　String exchange, String routingKey, 
　　　　　　　　　　　　　　　　AMQP.BasicProperties basicProperties, 
　　　　　　　　　　　　　　　　byte[] body) throws IOException { 
　　　　String message = new String(body); 
　　　　System.out.println("Basic.Return 返回的結果是: " + message);
　　}
}); 
.

 

二、AE備份交換器

　　Alternate Exchange，簡稱AE，不設置mandatory參數，那麼消息將會被丟失，設置mandatory參數的話，須要添加ReturnListner監聽器，增長複雜代碼，若是既不想增長代碼又不想消息丟失，則使用AE，將沒有被路由的消息存儲於RabbitMQ中。當mandatory參數用AE一塊兒使用時，mandatory將失效。在介紹AE以前，也認識RabbitMQ對於消息的過時時間TTL設置以及隊列的過時時間TTL設置

2.1 TTL過時時間設置

　　能夠對隊列設置TTL與消息設置TTL，其中消息設置TTL常常用於死信隊列、延遲隊列等高級應用中。

　　1）設置消息TTL

　　設置TTL過時時間通常有兩種當時：一是經過隊列屬性，對隊列中全部消息設置相同的TTL。二就是對消息自己單獨設置，每條消息TTL不一樣。若是一塊兒使用時候，TTL小的爲準，當一旦超過設置的TTL時間時，就會變成「死信」。

　　方式一：針對每條消息設置TTL是經過增長expiration的屬性參數實現的，不可能像方式二同樣掃描整個隊列再判斷是否過時，只有當該消息即將被消費時再斷定是否過時便可刪除，也就是消息即便已通過期，但不必定立馬被刪除！

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); 
// 持久化消息
builder deliveryMode(2);
// 設置 TTL=60000ms
builder expiration( 60000 ); 
AMQP.BasicProperties properties = builder. build(); 
channel.basicPublish(exchangeName, routingKey, mandatory, properties, "ttlTestMessage".getBytes());

　　方式二：經過隊列屬性設置消息TTL是增長x-message-ttl參數實現的，只須要掃描整個隊列頭部便可當即刪除，也就是消息一旦過時就會被刪除！

Map<String, Object> argss = new HashMap<String , Object>(); 
argss.put("x-message-ttl", 6000); 
channel.queueDeclare(queueName, durable, exclusive, autoDelete, argss) ;

 

　　2）設置隊列TTL

　　經過在隊列中添加參數x-message-ttl參數實現，設置隊列被自動刪除前處於未被使用狀態的時間，注意是隊列的使用狀態，並非消息是否被消費的狀態

　　設置ttl=30min的隊列，時間一到RabbitMQ會保證隊列被刪除，可是不會保證刪除的速度有多快。

Map<String, Object> args = new HashMap<String, Object>{); 
args.put("x-expires", 1800000);
channel.queueDeclare("myqueue", false, false, false, args);

 

 

2.2 AE備份交換器的使用

　　聲明交換器的時候，添加alternate-exchange參數實現，或經過策略實現。前者優先級高。從代碼角度須要如下三個步驟，具體代碼以下：

Map<String, Object> args = new HashMap<String, Object>(); 
args.put("a1ternate-exchange", "myAe"); 
channe1.exchangeDec1are("norma1Exchange", "direct", true, fa1se, args); 
channe1.exchangeDec1are("myAe", "fanout", true, fa1se, nu11) ; 
channe1.queueDec1are( "norma1Queue", true, fa1se, fa1se, nu11); 
channe1.queueB nd("norma1Queue", "norma1Exchange", "norma1Key"); 
channe1.queueDec1are("unroutedQueue", true, fa1se, fa1se, nu11);

　　1）聲明normalExchange類型爲direct的交換器、類型爲fanout的myAe備份交換器；而且normalExchange的備份交換器爲myAe（備份交換器建議使用fanout類型交換器）

　　2）聲明normalQueue隊列，聲明unrouteQueue隊列；

　　3）經過路由鍵normalKey綁定normalExchange與normalQueue，不適用路由鍵綁定unrouteQueue與myAe