前言

上一篇四種途徑提高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();
    // 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
}

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集群..........

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