消息中間件——RabbitMQ（七）高級特性全在這裡!（上）

2019 年 10 月 3 日
筆記

前言

前面我們介紹了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 查看管控台：

隊列1

隊列2

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 查看管控台

queues

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消息中間件精講》