訊息中介軟體——RabbitMQ（七）高階特性全在這裡!（上）

阿新 • • 發佈：2019-08-26

前言

前面我們介紹了RabbitMQ的安裝、各大訊息中介軟體的對比、AMQP核心概念、管控臺的使用、快速入門RabbitMQ。本章將介紹RabbitMQ的高階特性。分兩篇（上/下）進行介紹。

訊息如何保障100%的投遞成功？
冪等性概念詳解
在海量訂單產生的業務高峰期，如何避免訊息的重複消費的問題？
Confirm確認訊息、Return返回訊息

1 訊息如何保障100%的投遞成功？

1.1 什麼是生產端的可靠性投遞？

保障訊息的成功發出
保障MQ節點的成功接收
傳送端收到MQ節點（Broker）確認應答
完善的訊息進行補償機制

前三步不一定能保障訊息能夠100%投遞成功。因此要加上第四步

BAT/TMD 網際網路大廠的解決方案：
- 訊息落庫，對訊息狀態進行打標
在傳送訊息的時候，需要將訊息持久化到資料庫中，並給這個訊息設定一個狀態（未傳送、傳送中、到達）。當訊息狀態發生了變化，需要對訊息做一個變更。針對沒有到達的訊息做一個輪訓操作，重新發送。對輪訓次數也需要做一個限制3-5次。確保訊息能夠成功的傳送.

訊息的延遲投遞,做二次確認,回撥檢查

具體採用哪種方案,還需要根據業務與訊息的併發量而定。

1.2 第一種方案：

生產端-可靠性投遞

圖解：

藍色部分表示：生產者負責傳送訊息傳送至Broker端
Biz DB：訂單資料庫 MSG DB: 訊息資料
面對小規模的應用可以採用加事務的方式，保證事務的一致性。但在大廠中面對高併發，並沒有加事務，事務的效能拼接非常嚴重，而是做補償。

比如：如下發一條訂單訊息。

step1：儲存訂單訊息（建立訂單），業務資料入庫，訊息也入庫。缺點:需要持久化兩次。（status:0）
step2：在step1成功的前提下，傳送訊息
step3：Broker收到訊息後，confirm給我們的生產端。Confirm Listener非同步監聽Broker回送的訊息。
step4：抓取出指定的訊息，更新（status=1），表示訊息已經投遞成功。

step5：分散式定時任務獲取訊息狀態，如果等於0則抓取資料出來。
step6：重新發送訊息
step7：重試限制設定3次。如果訊息重試了3次還是失敗，那麼（status=2）,認為這個訊息就是失敗的。

查詢這些訊息為什麼失敗，可能需要人工去查詢。

假設step2執行成功，step3由於網路閃斷。那麼confirm將永遠收不到訊息，那麼我們需要設定一個規則：
例如：在訊息入庫的時候，設定一個臨界值 timeout=5min，當超過5min之後，就將這條資料抓取出來。
或者寫一個定時任務每隔5分鐘就將status=0的訊息抓取出來。可能存在小問題：訊息傳送出去，定時任務又正好剛執行，Confirm還未收到，定時任務就會執行，會導致訊息執行兩次。
更精細化操作：訊息超時容忍限制。confirm在2-3分鐘內未收到訊息，則重新發送。

保障MQ我們思考如果第一種可靠性投遞，在高併發的場景下是否合適？

第一種方案對資料有兩次入庫，一次業務資料入庫，一次訊息入庫。這樣對資料的入庫是一個瓶頸。
其實我們只需要對業務進行入庫。

訊息的延遲投遞，做二次確認，回撥檢查

這種方式並不一定能保證100%成功，但是也能保證99.99%的訊息成功。如果遇到特別極端的情況，那麼就只能需要人工去補償，或者定時任務去做。
第二種方式主要是為了減少對資料庫的操作。

看下第二種方式：

圖解：

Upstream service:生產端
DownStream service:消費端
Callback service:回撥服務

step1:業務訊息入庫成功後，第一次訊息傳送。
step2:同樣在訊息入庫成功後，傳送第二次訊息，這兩條訊息是同時傳送的。第二條訊息是延遲檢查，可以設定2min、5min 延遲傳送。
step3:消費端監聽指定佇列。
step4:消費端處理完訊息後，內部生成新的訊息send confirm。投遞到MQ Broker。
step5: Callback Service 回撥服務監聽MQ Broker,如果收到Downstream service傳送的訊息，則可以確定訊息傳送成功，執行訊息儲存到MSG DB。
step6：Check Detail檢查監聽step2延遲投遞的訊息。此時兩個監聽的佇列不是同一個，5分鐘後，Callback service收到訊息，檢查MSG DB。如果發現之前的訊息已經投遞成功，則不需要做其他事情。如果檢查發現失敗，則Callback 進行補償，主動傳送RPC 通訊。通知上游生產端重新發送訊息。

這樣做的目的：少做了一次DB儲存。關注點並不是百分百的投遞成功，而是效能。

2. 冪等性概念

2.1 冪等性是什麼？

冪等（idempotent、idempotence）是一個數學與計算機學概念，常見於抽象代數中，即f(f(x)) = f(x)。簡單的來說就是一個操作多次執行產生的結果與一次執行產生的結果一致。

我們可以借鑑資料庫的樂觀鎖機制：
比如我們執行一條更新庫存的SQL語句：
UPDATE T_REPS SET COUNT = COUNT - 1,VERSION = VERSION + 1 WHERE VERSION = 1

利用加版本號Version的方式來保證冪等性。

推薦文章：面試必備的資料庫悲觀鎖與樂觀鎖

2.2 消費端-冪等性保障

在海量訂單產生的業務高峰期，如何避免訊息的重複消費問題？

在高併發的情況下，會有大量的訊息到達MQ，消費端需要監聽大量的訊息。這樣的情況下，難免會出現訊息的重複投遞，網路閃斷等等。如果不去做冪等，則會出現訊息的重複消費。
-消費端實現冪等性，就意味著，我們的訊息永遠不會被消費多次，即使我們收到了多條一樣的訊息，也只會執行一次。

看下網際網路大廠主流的冪等性操作：
-唯一ID+指紋嗎機制，利用資料庫主鍵去重。
-利用Redis的原子性實現
-其他的技術實現冪等性

2.2.1 唯一ID+指紋碼機制

唯一ID + 指紋嗎機制，利用資料庫主鍵去重。
保證唯一性
SELECT COUNT(1) FROM T_ORDER WHERE ID = 唯一ID + 指紋碼
如果查詢沒有，則新增。有則不需要做任何操作，消費端不需要消費訊息。
好處：實現簡單
壞處：高併發下有資料庫寫入的效能瓶頸
解決方案：跟進ID進行分庫分表進行演算法路由
分攤流量壓力。

2.2.2 Redis 原子特性實現

最簡單使用Redis的自增。

使用Redis進行冪等，需要考慮的問題。
第一：我們是否需要資料落庫，如果落庫的話，關鍵解決的問題是資料庫和快取如何做到原子性？
加事務不行，Redis和資料庫的事務不是同一個，無法保證同時成功同時失敗。大家有什麼更好的方案呢？
第二：如果不進行落庫，那麼都儲存到快取中，如何設定定時同步的策略？
怎麼做到快取資料的穩定性？

3. Confirm 確認訊息

理解Confirm 訊息確認機制：

訊息的確認，是指生產者投遞訊息後，如果Broker收到訊息，則會給我們生產者一個應答。
生產者進行接收應答，用來確定這條訊息是否正常的傳送到Broker，這種方式也是訊息的可靠性投遞的核心保障！

藍色：producer 生產者紅色：MQ Broker 伺服器

生產者把訊息傳送到Broker端，Broker收到訊息之後回送給producer。Confirm Listener 監聽應答。

操作是非同步操作，當生產者傳送完訊息之後，就不需要管了。Confirm Listener 監聽MQ Broker的應答。

3.1 如何實現Confirm確認訊息？

第一步：在channel上開啟確認模式：channel.confirmSelect()
第二步；在chanel上新增監聽：addConfirmListener,監聽成功和失敗的返回結果，根據具體的結果對訊息進行重新發送、或記錄日誌等後續處理！

3.2 程式碼編寫：

生產者：


/**
 * 
* @ClassName: Producer 
* @Description: 生產者
* @author Coder程式設計
* @date 2019年7月30日 上午21:27:02 
*
 */
public class Producer {

    
    public static void main(String[] args) throws Exception {
        
        
        //1 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
    
        //2 通過Connection建立一個新的Channel
        Channel channel = connection.createChannel();
        
        
        //3 指定我們的訊息投遞模式: 訊息的確認模式 
        channel.confirmSelect();
        
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.save";
        
        //4 傳送一條訊息
        String msg = "Hello RabbitMQ Send confirm message!";
        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
        
        //5 新增一個確認監聽  用於傳送訊息到Broker端之後，回送訊息的監聽
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------no ack!-----------");
            }
            
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------ack!-----------");
            }
        });
    }
}

消費者：


/**
 * 
* @ClassName: Consumer 
* @Description: 消費者
* @author Coder程式設計
* @date 2019年7月30日 上午21:32:02 
*
 */
public class Consumer {

    
    public static void main(String[] args) throws Exception {
        
        
        //1 獲取一個連線 
        Connection connection = ConnectionUtils.getConnection();
        
        //2通過Connection建立一個新的Channel
        Channel channel = connection.createChannel();
        
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.#";
        String queueName = "test_confirm_queue";
        
        //3 宣告交換機和佇列 然後進行繫結設定, 最後制定路由Key
        channel.exchangeDeclare(exchangeName, "topic", true);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        //4 建立消費者 
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, true, queueingConsumer);
        
        while(true){
            Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            
            System.err.println("消費端: " + msg);
        }
        
        
    }
}

工具類：


/**
 * 
* @ClassName: ConnectionUtils 
* @Description: 連線工具類
* @author Coder程式設計
* @date 2019年6月21日 上午22:28:22 
*
 */
public class ConnectionUtils {
    public static Connection getConnection() throws IOException, TimeoutException {
        //定義連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設定服務地址
        factory.setHost("127.0.0.1");
        //埠
        factory.setPort(5672);//amqp協議 埠 類似與mysql的3306
        //設定賬號資訊，使用者名稱、密碼、vhost
        factory.setVirtualHost("/vhost_cp");
        factory.setUsername("user_cp");
        factory.setPassword("123456");
        // 通過工程獲取連線
        Connection connection = factory.newConnection();
        return connection;
    }
}

先啟動消費端=》再啟動生產端

3.3 檢視管控臺：

3.4 列印結果：

可以觀察到消費端先接收到訊息，之後生產端再接收到回撥資訊。如果出現磁碟已滿、RabbitMQ出現異常、queue容量到達上限都可能接收到no ack

如果ack和no ack訊息都未接收到，這就是之前所說的。RabbitMQ出現網路閃斷，可以採用上面所說的訊息補償。

4. Return訊息機制

Return Listener用於處理一些不可路由的訊息！
我們的訊息生產者，通過指定一個Exchange和Routingkey，把訊息送達到某一個佇列中去，然後我們的消費者監聽佇列，進行消費處理操作！
但是在某些情況下，如果我們在傳送訊息的時候，當前的exchange不存在或者指定的路由key路由不到，這個時候如果我們需要監聽這種不可達的訊息，就要使用Return Listener!

在基礎API中有一個關鍵的配置項：

Mandatory:如果為true，則監聽器會接收到路由不可達的訊息，然後進行後續處理，如果為false,那麼broker端自動刪除該訊息！

4.1 Return訊息機制流程

Producer生產端將訊息傳送到MQ Broker端，但是出現NotFind Exchange,傳送的訊息的Exchange，在Broker端未能找到。或者找到了，但是路由key路由不到指定的佇列。因此是一個錯誤的訊息。
這個時候，生產端應該知道傳送的這條訊息，並不會被處理。因此MQ Broker提供了這種Return機制，將這些不可達的訊息傳送給生產端，這時候生產端就需要設定Return Listener去接收這些不可達的訊息。然後及時記錄日誌，去處理這些訊息。

4.2 程式碼演示

生產者：


/**
 * 
* @ClassName: Producer 
* @Description: 生產者
* @author Coder程式設計
* @date 2019年7月30日 上午22:03:22 
*
 */
public class Producer {

    
    public static void main(String[] args) throws Exception {
        
        
        //1 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        
        Channel channel = connection.createChannel();
        
        String exchange = "test_return_exchange";
        String routingKey = "return.save";
        String routingKeyError = "abc.save";
        
        String msg = "Hello RabbitMQ Return Message";
        
        
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange,
                    String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                
                System.err.println("---------handle  return----------");
                //響應碼
                System.err.println("replyCode: " + replyCode);
                //響應文字
                System.err.println("replyText: " + replyText);
                System.err.println("exchange: " + exchange);
                System.err.println("routingKey: " + routingKey);
                System.err.println("properties: " + properties);
                System.err.println("body: " + new String(body));
            }
        });
        
        //第三個引數mandatory=true,意味著路由不到的話mq也不會刪除訊息,false則會自動刪除
        channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        //修改routingkey，測試是否能夠收到訊息
        //channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
    }
}

消費者：


/**
 * 
* @ClassName: Consumer 
* @Description: 消費者
* @author Coder程式設計
* @date 2019年7月30日 上午22:33:34 
*
 */
public class Consumer {

    
    public static void main(String[] args) throws Exception {
        
        
        //1 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        
        Channel channel = connection.createChannel();
        
        String exchangeName = "test_return_exchange";
        String routingKey = "return.#";
        String queueName = "test_return_queue";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        
        channel.basicConsume(queueName, true, queueingConsumer);
        
        while(true){
            Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.err.println("消費者: " + msg);
        }
    }
}

ConnectionUtils 工具程式碼在上面。

啟動消費端，並檢視管控臺。

4.3 檢視管控臺

4.4 檢視列印結果

放開消費端程式碼：channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
消費端列印結果：

可以看到列印結果正常，此時再改程式碼為：
channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());

可以看到生產端接收到了不可達的訊息。

文末

歡迎關注個人微信公眾號：Coder程式設計
獲取最新原創技術文章和免費學習資料，更有大量精品思維導圖、面試資料、PMP備考資料等你來領，方便你隨時隨地學習技術知識！
新建了一個qq群：315211365，歡迎大家進群交流一起學習。謝謝了！也可以介紹給身邊有需要的朋友。

文章收錄至
Github: https://github.com/CoderMerlin/coder-programming
Gitee: https://gitee.com/573059382/coder-programming
歡迎關注並star~

參考文章：

《RabbitMQ訊息中介軟體精講》