四種途徑提高RabbitMQ傳輸資料的可靠性（二）
前言
上一篇 四種途徑提高RabbitMQ傳輸訊息資料的可靠性（一） 已經介紹了兩種方式提高資料可靠性傳輸的方法，本篇針對上一篇中提出的問題（1）與問題（2）提出解決常用的方法。
本文其實也就是結合以上四個方面進行講解的，主要參考《RabbitMQ實戰指南》（有需要PDF電子書的可以評論或者私信我），本文截圖也來自其中，另外可以對一些RabbitMQ的概念的認識可以參考我的另外兩篇博文 認識RabbitMQ交換機模型 、 RabbitMQ是如何運轉的？
三、生產者確認機制
針對問題（1），我們可以通過生產者的確認訊息機制來解決，主要分為兩種：第一是事務機制、第二是傳送方確認機制
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();
發生異常之後事務回滾
try {
channel.txSelect();
channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, "transaction messages".getBytes());
channel.txCommit();
} catch (Exception e){
e.printStackTrace();
channel.txRollback();
}
2、確認機制
生產者將通道設定為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();
}
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();
// takeas 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
}
3、總結比較
1）是確認機制好呢？還是事務機制？兩者可以共存嗎？
確認機制相對於事務機制，最大的好處就是可以非同步處理提高吞吐量，不需要額外等待消耗資源。但是兩者時候不能同時共存的。
2）那麼確認機制的三種方式之間呢？實際產生環境是推薦哪一種呢？（其實毫無疑問當然是推薦非同步批量確認方式）
批量確認的最大問題就是在於返回的Nack訊息需要重新發送，以上普通單條確認、批量確認、批量非同步確認三種方法，在實際生產環境中強烈推薦使用批量非同步確認方式。
四、訊息與佇列的持久化
針對的問題（2），我們可以通過增加佇列與訊息的持久化來實現。
1、交換器的持久化
交換器的持久化是通過宣告佇列durable引數為true實現的， 如果交換器不設定持久化，那麼在RabbitMQ伺服器重啟之後，相關的交換器元資料會丟失，訊息不會丟失，只是不能將訊息傳送到這個交換器中 。因此，都是建議將其置為持久化。
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
2、佇列的持久化
佇列持久化同理與交換器持久化，只是RabbitMQ伺服器重啟之後，相關的元資料會丟失，資料也會跟著丟失，訊息也自然丟失。
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
3、訊息的持久化
佇列的持久化不能保證記憶體儲存的訊息不會丟失， 要確保訊息不會丟失，需要將其通過設定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);
4、訊息丟失的幾種情況
但實際上不是設定了交換器、佇列、訊息持久化就能一定保證訊息不會被丟失，以下幾種情況是可能丟失的，比如：
1） 設定autoAck為true，消費者收到訊息後，還沒處理就宕機了，這樣也算資料丟失，解決辦法是設定為false，之後手動確認。
2）在設定了持久化後訊息存入RabbitMQ之後，還需要一段時間才能存入磁碟之中（雖然很短，但不能忽視），RabbitMQ並不會為每條訊息都今次那個同步存檔，可能只會儲存到作業系統快取之中而不是物理磁碟中 ， 如果RabbitMQ這個時間段內宕機、異常、重啟等情況，訊息也會丟失，解決辦法是引入RabbitMQ的映象佇列機制（類似於叢集，Master掛了切換到Slave）
總結
沒有完全十全十美的方式能保證資料能100%不丟失，並且最大效率節約效能消耗等，兩篇博文雖然已經提出常用的四種方式，當實際環境中整個RabbitMQ環境在搭建沒有結合實際的生產業務環境的話，也會發生訊息丟失的等情況，解決這樣的問題無非就完善訊息備份，健全RabbitMQ叢集..........