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

  • 2019 年 10 月 3 日
  • 筆記

求關注

高級特性全在這裡!（下）

前言

上一篇消息中間件——RabbitMQ（七）高級特性全在這裡!（上）中我們介紹了消息如何保障100%的投遞成功？,冪等性概念詳解,在海量訂單產生的業務高峰期，如何避免消息的重複消費的問題？,Confirm確認消息、Return返回消息。這篇我們來介紹下下面內容。

  • 自定義消費者
  • 消息的限流（防止佔用記憶體過多，節點宕機）
  • 消息的ACK與重回隊列
  • TTL消息
  • 死信隊列

1. 自定義消費者

1.1 消費端自定義監聽

我們一般就在程式碼中編寫while循環，進行consumer.nextDelivery方法進行獲取下一條消息，然後進行消費處理！

但是這種輪訓的方式肯定是不好的，程式碼也比較low。

  • 我們使用自定義的Consumer更加的方便，解耦性更加的強，也是在實際工作中最常見的使用方式！

消費端自定義監聽

1.2 程式碼演示

1.2.1 生產者

  /**   *  * @ClassName: Producer  * @Description: 生產者  * @author Coder編程  * @date2019年7月30日 下午23:15:51  *   */  public class Producer {          public static void main(String[] args) throws Exception {            //1 創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();            String exchange = "test_consumer_exchange";          String routingKey = "consumer.save";            String msg = "Hello RabbitMQ Consumer Message";            for(int i =0; i<5; i ++){              channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());          }        }  }

1.2.2 消費者

  /**   *  * @ClassName: Consumer  * @Description: 消費者  * @author Coder編程  * @date2019年7月30日 下午23:13:51  *   */  public class Consumer {          public static void main(String[] args) throws Exception {              // 創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();              String exchangeName = "test_consumer_exchange";          String routingKey = "consumer.#";          String queueName = "test_consumer_queue";            channel.exchangeDeclare(exchangeName, "topic", true, false, null);          channel.queueDeclare(queueName, true, false, false, null);          channel.queueBind(queueName, exchangeName, routingKey);            //實現自己的MyConsumer()          channel.basicConsume(queueName, true, new MyConsumer(channel));      }  }

1.2.3 自定義類：MyConsumer

/**   *  * @ClassName: MyConsumer  * @Description: TODO  * @author Coder編程  * @date 2019年7月30日 下午23:11:55  *   */  public class MyConsumer extends DefaultConsumer {          public MyConsumer(Channel channel) {          super(channel);      }        //根據需求，重寫自己需要的方法。      @Override      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {          System.err.println("-----------consume message----------");          //消費標籤          System.err.println("consumerTag: " + consumerTag);          //這個對象包含許多關鍵資訊          System.err.println("envelope: " + envelope);          System.err.println("properties: " + properties);          System.err.println("body: " + new String(body));      }      }

1.3 列印結果

列印結果

2. 消費端限流

2.1 什麼是消費端的限流？

  • 假設一個場景，首先，我們Rabbitmq伺服器有上萬條未處理的消息，我們隨便打開一個消費者客戶端，會出現下面情況：
  • 巨量的消息瞬間全部推送過來，但是我們單個客戶端無法同時處理這麼多數據！這個時候很容易導致伺服器崩潰，出現故障。

為什麼不在生產端進行限流呢？

因為在高並發的情況下，客戶量就是非常大，所以很難在生產端做限制。因此我們可以用MQ在消費端做限流。

  • RabbitMQ提供了一種qos(服務品質保證)功能，即在非自動確認消息的前提下，如果一定數目的消息（通過基於consume或者channel設置Qos的值）未被確認前，不進行消費新的消息。
    在限流的情況下，千萬不要設置自動簽收，要設置為手動簽收。
  • void BasicQos(uint prfetchSize,ushort prefetchCount,bool global);

參數解釋：
prefetchSize:0
prefetchCount:會告訴RabbitMQ不要同時給一個消費者推送多於N個消息，即一旦有N個消息還沒有ack，則該consumer將block掉，直到有消息ack。
global: truefalse 是否將上面設置應用於channel，簡單點說，就是上面限制是channel級別還是consumer級別。
prefetchSize和global這兩項，rabbitmq沒有實現，暫且不研究prefetch_count在no_ask = false的情況下生效，即在自動應答的情況下這兩個值是不生效的。

第一個參數：消息的限制大小，消息多少兆。一般不做限制，設置為0
第二個參數：一次最多處理多少條，實際工作中設置為1就好
第三個參數：限流策略在什麼上應用。在RabbitMQ一般有兩個應用級別：1.通道 2.Consumer級別。一般設置為false，true 表示channel級別，false表示在consumer級別

2.2 程式碼演示

2.2.1 生產者

  /**   *  * @ClassName: Producer  * @Description: 生產者  * @author Coder編程  * @date2019年7月30日 下午23:15:51  *   */  public class Producer {          public static void main(String[] args) throws Exception {            //1 創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();            String exchange = "test_qos_exchange";          String routingKey = "qos.save";            String msg = "Hello RabbitMQ QOS Message";            for(int i =0; i<5; i ++){              channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());          }        }  }

2.2.2 消費者

  /**   *  * @ClassName: Consumer  * @Description: 消費者  * @author Coder編程  * @date2019年7月30日 下午23:13:51  *   */  public class Consumer {          public static void main(String[] args) throws Exception {              //1 創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();              String exchangeName = "test_qos_exchange";          String queueName = "test_qos_queue";          String routingKey = "qos.#";            channel.exchangeDeclare(exchangeName, "topic", true, false, null);          channel.queueDeclare(queueName, true, false, false, null);          channel.queueBind(queueName, exchangeName, routingKey);            //1 限流方式  第一件事就是 autoAck設置為 false          //設置為1，表示一條一條數據處理          channel.basicQos(0, 1, false);            channel.basicConsume(queueName, false, new MyConsumer(channel));          }  }

2.2.3 自定義類：MyConsumer

  /**   *  * @ClassName: MyConsumer  * @Description: TODO  * @author Coder編程  * @date 2019年7月30日 下午23:11:55  *   */  public class MyConsumer extends DefaultConsumer {          private Channel channel ;        public MyConsumer(Channel channel) {          super(channel);          this.channel = channel;      }        @Override      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {          System.err.println("-----------consume message----------");          System.err.println("consumerTag: " + consumerTag);          System.err.println("envelope: " + envelope);          System.err.println("properties: " + properties);          System.err.println("body: " + new String(body));            //需要做簽收,false表示不支援批量簽收          channel.basicAck(envelope.getDeliveryTag(), false);        }      }

2.2.4 測試結果

我們先注釋掉：channel.basicAck(envelope.getDeliveryTag(), false);然後啟動Consumer。
查看Exchange

Exchange

查看Queues

Queues

然後再啟動Producer。查看列印結果：

列印結果

我們會發現消費端，只收到了一條消息。這是為什麼呢？

第一點因為我們在consumer中

channel.basicConsume(queueName, false, new MyConsumer(channel));

第二個參數設置為false為手動簽收。

第二點在qos中設置只接受一條消息。如果這一條消息不給Broker Ack應答的話，那麼Broker會認為你並沒有消費完這一條消息，那麼就不會繼續發送消息。

channel.basicQos(0, 1, false);

可以看下管控台，unack=1，Ready=4，total=5.

queues

接下來我們放開注釋channel.basicAck(envelope.getDeliveryTag(), false); 進行消息簽收。重啟服務。

3.1 列印結果

列印結果

可以看到正常列印五條結果

4. 消費端ACK與重回隊列

4.1 消費端的手工ACK和NACK

消費端進行消費的時候，如果由於業務異常我們可以進行日誌的記錄，然後進行補償！

如果由於伺服器宕機等嚴重問題，那我們就需要手工進行ACK保障消費端消費成功！

4.2 消費端的重回隊列

消費端重回隊列是為了對沒有處理成功的消息，把消息重新傳遞給Broker!

一般我們在實際應用中，都會關閉重回隊列，也就是設置為False.

4.3 程式碼演示

4.3.1 生產者

  /**   *  * @ClassName: Producer  * @Description: 生產者  * @author Coder編程  * @date2019年7月30日 下午23:15:51  *   */  public class Producer {          public static void main(String[] args) throws Exception {            //1創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();            String exchange = "test_ack_exchange";          String routingKey = "ack.save";                for(int i =0; i<5; i ++){                Map<String, Object> headers = new HashMap<String, Object>();              headers.put("num", i);                //添加屬性，後續會使用到              AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()                      .deliveryMode(2) //投遞模式，持久化                      .contentEncoding("UTF-8")                      .headers(headers)                      .build();              String msg = "Hello RabbitMQ ACK Message " + i;              channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());          }        }  }

4.3.2 消費者

  /**   *  * @ClassName: Consumer  * @Description: 消費者  * @author Coder編程  * @date2019年7月30日 下午23:13:51  *   */  public class Consumer {          public static void main(String[] args) throws Exception {              //1創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();              String exchangeName = "test_ack_exchange";          String queueName = "test_ack_queue";          String routingKey = "ack.#";            channel.exchangeDeclare(exchangeName, "topic", true, false, null);          channel.queueDeclare(queueName, true, false, false, null);          channel.queueBind(queueName, exchangeName, routingKey);            // 手工簽收 必須要關閉 autoAck = false          channel.basicConsume(queueName, false, new MyConsumer(channel));          }  }

4.3.3 自定義類：MyConsumer

/**   *  * @ClassName: MyConsumer  * @Description: TODO  * @author Coder編程  * @date 2019年7月30日 下午23:11:55  *   */  public class MyConsumer extends DefaultConsumer {          private Channel channel ;        public MyConsumer(Channel channel) {          super(channel);          this.channel = channel;      }        @Override      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {          System.err.println("-----------consume message----------");          System.err.println("body: " + new String(body));          try {              Thread.sleep(2000);          } catch (InterruptedException e) {              e.printStackTrace();          }          if((Integer)properties.getHeaders().get("num") == 0) {              //Nack三個參數  第二個參數：是否是批量，第三個參數：是否重回隊列（需要注意可能會發生重複消費，造成死循環）              channel.basicNack(envelope.getDeliveryTag(), false, true);          } else {              channel.basicAck(envelope.getDeliveryTag(), false);          }        }      }

5.1 列印結果：

列印結果

注意：
可以看到重回隊列會出現重複消費導致死循環的問題，這時候最好設置重試次數，比如超過三次後，消息還是消費失敗，就將消息丟棄。

6. TTL隊列/消息

6.1 TTL

  • TTL是Time To Live的縮寫，也就是生存時間
  • RabbitMQ支援消息的過期時間，在消息發送時可以進行指定
  • RabbitMQ支援隊列的過期時間，從消息入隊列開始計算，只要超過了隊列的超時時間配置，那麼消息會自動的清除

6.2 程式碼演示

6.2.1 直接通過管控台進行演示

添加queue

通過管控台創建一個隊列

x-max-length 隊列的最大大小
x-message-ttl 設置10秒鐘，如果消息還沒有被消費的話，就會被清除。

queue添加成功

添加exchange

添加exchange

Queue與Exchange進行綁定

點擊 test_ttl_exchange 進行綁定

進行綁定

查看是否綁定成功

綁定1

綁定2

通過管控台發送消息

發送消息

queue中有消息

消息未處理自動清除
十秒後消息消失

生產端設置過期時間

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()                  .deliveryMode(2)                  .contentEncoding("UTF-8")                  .expiration("10000")                  .headers(headers)                  .build();

這兩個屬性並不相同，一個對應的是消息體，一個對應的是隊列的過期。

7. 死信隊列

7.1 概念理解

死信隊列：DLX,Dead-Letter-Exchange
RabbitMQ的死信隊里與Exchange息息相關

  • 利用DLX,當消息在一個隊列中變成死信（dead message）之後，它能被重新publish到另一個Exchange，這個Exchange就是DLX

消息變成死信有以下幾種情況

  • 消息被拒絕（basic.reject/basic.nack）並且requeue=false
  • 消息TTL過期
  • 隊列達到最大長度

DLX也是一個正常的Exchange，和一般的Exchange沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性

當這個隊列中有死信時，RabbitMQ就會自動的將這個消息重新發布到設置的Exchange上去，進而被路由到另一個隊列。

可以監聽這個隊列中消息做相應的處理，這個特徵可以彌補RabbitMQ3.0以前支援的immediate參數的功能。

7.2 程式碼演示

  • 死信隊列設置：
  • 首先需要設置死信隊列的exchange和queue,然後進行綁定：
    Exchange:dlx.exchange
    Queue:dlx.queue
    RoutingKey:#
  • 然後我們進行正常聲明交換機、隊列、綁定，只不過我們需要在隊列加上一個參數即可：arguments.put("x-dead-letter-exchange","dlx.exchange");
  • 這樣消息在過期、requeue、隊列在達到最大長度時，消息就可以直接路由到死信隊列！

7.2.1 生產者

/**   *  * @ClassName: Producer  * @Description: 生產者  * @author Coder編程  * @date2019年7月30日 下午23:15:51  *   */  public class Producer {          public static void main(String[] args) throws Exception {            //創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();            String exchange = "test_dlx_exchange";          String routingKey = "dlx.save";            String msg = "Hello RabbitMQ DLX Message";            for(int i =0; i<1; i ++){                AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()                      .deliveryMode(2)                      .contentEncoding("UTF-8")                      .expiration("10000")                      .build();              channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());          }        }  }

7.2.2 消費者

/**   *  * @ClassName: Consumer  * @Description: 消費者  * @author Coder編程  * @date2019年7月30日 下午23:13:51  *   */  public class Consumer {          public static void main(String[] args) throws Exception {              //創建ConnectionFactory          Connection connection = ConnectionUtils.getConnection();          Channel channel = connection.createChannel();            // 這就是一個普通的交換機 和 隊列 以及路由          String exchangeName = "test_dlx_exchange";          String routingKey = "dlx.#";          String queueName = "test_dlx_queue";            channel.exchangeDeclare(exchangeName, "topic", true, false, null);            Map<String, Object> agruments = new HashMap<String, Object>();          agruments.put("x-dead-letter-exchange", "dlx.exchange");          //這個agruments屬性，要設置到聲明隊列上          channel.queueDeclare(queueName, true, false, false, agruments);          channel.queueBind(queueName, exchangeName, routingKey);            //要進行死信隊列的聲明:          channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);          channel.queueDeclare("dlx.queue", true, false, false, null);          channel.queueBind("dlx.queue", "dlx.exchange", "#");            channel.basicConsume(queueName, true, new MyConsumer(channel));          }  }

7.2.3 自定義類：MyConsumer

  /**   *  * @ClassName: MyConsumer  * @Description: TODO  * @author Coder編程  * @date 2019年7月30日 下午23:11:55  *   */  public class MyConsumer extends DefaultConsumer {          public MyConsumer(Channel channel) {          super(channel);      }        @Override      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {          System.err.println("-----------consume message----------");          System.err.println("consumerTag: " + consumerTag);          System.err.println("envelope: " + envelope);          System.err.println("properties: " + properties);          System.err.println("body: " + new String(body));      }  }

7.2.4 測試結果

運行Consumer，查看管控台
查看Exchanges

exchange

查看queue

queue

可以看到test_dlx_queue多了DLX的標識，表示當隊列中出現死信的時候，會將消息發送到死信隊列dlx_queue中

關閉Consumer，只運行Producer

查看queue10秒鐘前

過10秒鐘後，消息過期
查看queue10秒鐘後

在我們工作中，死信隊列非常重要，用於消息沒有消費者，處於死信狀態。我們可以才用補償機制。

小結

本次主要介紹了RabbitMQ的高級特性，首先介紹了互聯網大廠在實際使用中如何保障100%的消息投遞成功和冪等性的，以及對RabbitMQ的確認消息、返回消息、ACK與重回隊列、消息的限流，以及對超時時間、死信隊列的使用

文末

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

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

參考文章：

《RabbitMQ消息中間件精講》

推薦文章：

消息中間件——RabbitMQ（五）快速入門生產者與消費者，SpringBoot整合RabbitMQ！

消息中間件——RabbitMQ（六）理解Exchange交換機核心概念!

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

VirMach 便宜 VPS

QNews

QNews