四種途徑提升RabbitMQ傳輸數據的可靠性（二）

3、生產者確認機制

針對問題（1），咱們能夠經過生產者的確認消息機制來解決，主要分爲兩種：第一是事務機制、第二是發送方確認機制

一、事務機制

與事務機制相關的有三種方法，分別是channel.txSelect設置當前信道爲事務模式、channel.txCommit提交事務和channel.txRollback事務回滾。若是事務提交成功，則消息必定是到達了RabbitMQ中，若是事務提交以前因爲發送異常或者其餘緣由，捕獲後能夠進行channel.txRollback回滾。

// 將信道設置爲事務模式，開啓事務
channel.txSelect(); 
// 發送持久化消息
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, "transaction messages".getBytes()); 
// 事務提交
channel.txCommit();

發生異常以後事務回滾

若是想學習Java工程化、高性能及分佈式、深刻淺出。微服務、Spring，MyBatis，Netty源碼分析的朋友能夠加個人Java高級交流：854630135，羣裏有阿里大牛直播講解技術，以及Java大型互聯網技術的視頻免費分享給你們。

try {
 　　channel.txSelect();
 channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, "transaction messages".getBytes());
 channel.txCommit();
　　} catch (Exception e){
 e.printStackTrace();
 channel.txRollback();
 }

二、確認機制

　生產者將信道設置爲confirm確認模式，確認以後全部在信道上的消息將會被指派一個惟一的從1開始的ID，一旦消息被正確匹配到全部隊列後，RabbitMQ就會發送一個確認Basic.Ack給生產者（包含消息的惟一ID），生產者便知曉消息是否正確到達目的地了。

 

 

消息若是是持久化的，那麼確認消息會在消息寫入磁盤以後發出。RabbitMQ中的deliveryTag包含了確認消息序號，還能夠設置multiple參數，表示到這個序號以前的全部消息都已經獲得處理。確認機制相對事務機制來講，相比較代碼來講比較複雜，但會常用，主要有單條確認、批量確認、異步批量確認三種方式。

2.1 單條確認

此種方式比較簡單，通常都是一條條的發送，代碼以下：

try {
 Connection connection = connectionFactory.newConnection();
 Channel channel = connection.createChannel();
 //set channel publisher confirm mode 
 channel.confirmSelect();
 // publish message
 channel.basicPublish("exchange", "routingkey", null, "publisher confirm test".getBytes());
 if (!channel.waitForConfirms()) {
 // publisher confirm failed handle
 System.out.println("send message failed!");
 }
} catch (Exception e) {
 e.printStackTrace();
}

若是想學習Java工程化、高性能及分佈式、深刻淺出。微服務、Spring，MyBatis，Netty源碼分析的朋友能夠加個人Java高級交流：854630135，羣裏有阿里大牛直播講解技術，以及Java大型互聯網技術的視頻免費分享給你們。

2.2 批量確認

問：批量確認comfirm須要解決出現返回的Basic.Nack或者超時狀況的話，客戶須要將這一批次消息所有重發，那麼採用什麼樣的存儲結構才能合適地將這些消息動態篩選出來。

最好是須要增長一個緩存，將發送成功而且確認Ack以後的消息去除，剩下Nack或者超時的消息，改進以後的代碼以下：

// take ArrayList or BlockingQueue as a cache
List<Object> cache = new ArrayList<>();
// set channel publisher confirm mode
channel.confirmSelect();
for (int i=0; i < 20; i++) {
 // publish message
 String message = "publisher message["+ i +"]";
 cache.add(message);
 channel.basicPublish("exchange", "routingkey", null, message.getBytes());
 if (channel.waitForConfirms()) {
 // remove message publisher confirm
 cache.remove(i);
 }
 // TODO handle Nack message：republish
}
} catch (Exception e) {
　　e.printStackTrace();
　　// TODO handle Nack message：republish
}

2.3 異步批量確認

異步確認方式經過在客戶端addConfirmListener增長ConfirmListener回調接口，包括handleAck與handleNack處理方法：

每次發送消息confirmSet集合元素加1，當消息被確認ack進入handleAck方法時，「unconfirm」集合中刪除響應的一條（multiple設置爲false時）或者多條記錄（multiple設置爲true時），其中存儲緩存最好採用SortedSet數據結構

代碼以下：

try {
 Connection connection = connectionFactory.newConnection();
 Channel channel = connection.createChannel();
 // take as a cache
 SortedSet cache = new TreeSet();
 // set channel publisher confirm mode
 channel.confirmSelect();
 for (int i = 0; i < 20; i++) {
 // publish message
 long nextSeqNo = channel.getNextPublishSeqNo();
 String message = "publisher message[" + i + "]";
 cache.add(message);
 channel.basicPublish("exchange", "routingkey", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
 cache.add(nextSeqNo);
 }
 // add confirmCalback: handleAck, handleNack
 channel.addConfirmListener(new ConfirmListener() {
 @Override
 public void handleAck(long deliveryTag, boolean multiple) {
 if (multiple) {
 // batch remove ack message
 cache.headSet(deliveryTag - 1).clear();
 } else {
 // remove ack message
 cache.remove(deliveryTag);
 }
 }
 @Override
 public void handleNack(long deliveryTag, boolean multiple) {
 // TODO handle Nack message：republish
 }
 });
} catch (Exception e) {
 e.printStackTrace();
 // TODO handle Nack message：republish
}

三、總結比較

1）是確認機制好呢？仍是事務機制？二者能夠共存嗎？

確認機制相對於事務機制，最大的好處就是能夠異步處理提升吞吐量，不須要額外等待消耗資源。可是二者時候不能同時共存的。

2）那麼確認機制的三種方式之間呢？實際產生環境是推薦哪種呢？（其實毫無疑問固然是推薦異步批量確認方式）

批量確認的最大問題就是在於返回的Nack消息須要從新發送，以上普通單條確認、批量確認、批量異步確認三種方法，在實際生產環境中強烈推薦使用批量異步確認方式。

4、消息與隊列的持久化

針對的問題（2），咱們能夠經過增長隊列與消息的持久化來實現。

一、交換器的持久化

交換器的持久化是經過聲明隊列durable參數爲true實現的，若是交換器不設置持久化，那麼在RabbitMQ服務器重啓以後，相關的交換器元數據會丟失，消息不會丟失，只是不能將消息發送到這個交換器中。所以，都是建議將其置爲持久化。

channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);

二、隊列的持久化

隊列持久化同理與交換器持久化，只是RabbitMQ服務器重啓以後，相關的元數據會丟失，數據也會跟着丟失，消息也天然丟失。

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

三、消息的持久化

隊列的持久化不能保證內存存儲的消息不會丟失，要確保消息不會丟失，須要將其經過設置BasicProperties中的deliveryMode屬性爲2可實現消息的持久化（PERSISTENT_TEXT_PLAIN實際上已經封裝了這個屬性），也就是說只有實現了隊列與消息的持久化，才能保證消息不會丟失。

// 其中的2就是投遞模式
public static Class final BasicProperties_PERSISTENT_TEXT_PLAIN = 
new BasicProperties("text/plain", null, null, 2, null, null, null, null, null, null, null, null, null);

四、消息丟失的幾種狀況

但實際上不是設置了交換器、隊列、消息持久化就能必定保證消息不會被丟失，如下幾種狀況是可能丟失的，好比：

1）設置autoAck爲true，消費者收到消息後，還沒處理就宕機了，這樣也算數據丟失，解決辦法是設置爲false，以後手動確認。

2）在設置了持久化後消息存入RabbitMQ以後，還須要一段時間才能存入磁盤之中（雖然很短，但不能忽視），RabbitMQ並不會爲每條消息都今次那個同步存盤，可能只會保存到操做系統緩存之中而不是物理磁盤中，若是RabbitMQ這個時間段內宕機、異常、重啓等狀況，消息也會丟失，解決辦法是引入RabbitMQ的鏡像隊列機制（相似於集羣，Master掛了切換到Slave）

　

總結

沒有徹底十全十美的方式能保證數據能100%不丟失，而且最大效率節約性能消耗等，兩篇博文雖然已經提出經常使用的四種方式，當實際環境中整個RabbitMQ環境在搭建沒有結合實際的生產業務環境的話，也會發生消息丟失的等狀況，解決這樣的問題無非就完善消息備份，健全RabbitMQ集羣..........

若是想學習Java工程化、高性能及分佈式、深刻淺出。微服務、Spring，MyBatis，Netty源碼分析的朋友能夠加個人Java高級交流：854630135，羣裏有阿里大牛直播講解技術，以及Java大型互聯網技術的視頻免費分享給你們。