目錄

• 消費端限流
• TTL
• 死信佇列

消費端限流

1. 為什麼要對消費端限流

假設一個場景，首先，我們 Rabbitmq 伺服器積壓了有上萬條未處理的訊息，我們隨便開啟一個消費者客戶端，會出現這樣情況: 巨量的訊息瞬間全部推送過來，但是我們單個客戶端無法同時處理這麼多資料!

當資料量特別大的時候，我們對生產端限流肯定是不科學的，因為有時候併發量就是特別大，有時候併發量又特別少，我們無法約束生產端，這是使用者的行為。所以我們應該對消費端限流，用於保持消費端的穩定，當訊息數量激增的時候很有可能造成資源耗盡，以及影響服務的效能，導致系統的卡頓甚至直接崩潰。

2.限流的 api 講解

RabbitMQ 提供了一種 qos （服務質量保證）功能，即在非自動確認訊息的前提下，如果一定數目的訊息（通過基於 consume 或者 channel 設定 Qos 的值）未被確認前，不進行消費新的訊息。

/**
* Request specific "quality of service" settings.
* These settings impose limits on the amount of data the server
* will deliver to consumers before requiring acknowledgements.
* Thus they provide a means of consumer-initiated flow control.
* @param prefetchSize maximum amount of content (measured in
* octets) that the server will deliver, 0 if unlimited
* @param prefetchCount maximum number of messages that the server
* will deliver, 0 if unlimited
* @param global true if the settings should be applied to the
* entire channel rather than each consumer
* @throws java.io.IOException if an error is encountered
*/
void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
 

• prefetchSize：0，單條訊息大小限制，0代表不限制

• prefetchCount：一次性消費的訊息數量。會告訴 RabbitMQ 不要同時給一個消費者推送多於 N 個訊息，即一旦有 N 個訊息還沒有 ack，則該 consumer 將 block 掉，直到有訊息 ack。

• global：true、false 是否將上面設定應用於 channel，簡單點說，就是上面限制是 channel 級別的還是 consumer 級別。當我們設定為 false 的時候生效，設定為 true 的時候沒有了限流功能，因為 channel 級別尚未實現。

• 注意：prefetchSize 和 global 這兩項，rabbitmq 沒有實現，暫且不研究。特別注意一點，prefetchCount 在 no_ask=false 的情況下才生效，即在自動應答的情況下這兩個值是不生效的。

3.如何對消費端進行限流

• 首先第一步，我們既然要使用消費端限流，我們需要關閉自動 ack，將 autoAck 設定為 falsechannel.basicConsume(queueName, false, consumer);

• 第二步我們來設定具體的限流大小以及數量。channel.basicQos(0, 15, false);

• 第三步在消費者的 handleDelivery 消費方法中手動 ack，並且設定批量處理 ack 迴應為 truechannel.basicAck(envelope.getDeliveryTag(), true);

這是生產端程式碼，與前幾章的生產端程式碼沒有做任何改變，主要的操作集中在消費端。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class QosProducer {
    public static void main(String[] args) throws Exception {
        //1. 建立一個 ConnectionFactory 並進行設定
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");

        //2. 通過連線工廠來建立連線
        Connection connection = factory.newConnection();

        //3. 通過 Connection 來建立 Channel
        Channel channel = connection.createChannel();

        //4. 宣告
        String exchangeName = "test_qos_exchange";
        String routingKey = "item.add";

        //5. 傳送
        String msg = "this is qos msg";
        for (int i = 0; i < 10; i++) {
            String tem = msg + " : " + i;
            channel.basicPublish(exchangeName, routingKey, null, tem.getBytes());
            System.out.println("Send message : " + tem);
        }

        //6. 關閉連線
        channel.close();
        connection.close();
    }


}

這裡我們建立了兩個消費者，以方便驗證限流api中的 global 引數設定為 true 時不起作用.。整體結構如下圖所示，兩個 Consumer 都繫結在同一個佇列上，這樣的話兩個消費者將共同消費傳送的10條訊息。

import com.rabbitmq.client.*;
import java.io.IOException;
public class QosConsumer {
    public static void main(String[] args) throws Exception {
        //1. 建立一個 ConnectionFactory 並進行設定
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(3000);

        //2. 通過連線工廠來建立連線
        Connection connection = factory.newConnection();

        //3. 通過 Connection 來建立 Channel
        final Channel channel = connection.createChannel();

        //4. 宣告
        String exchangeName = "test_qos_exchange";
        String queueName = "test_qos_queue";
        String queueName1 = "test_qos_queue_1";
        String routingKey = "item.#";
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);

        channel.basicQos(0, 3, false);

        //一般不用程式碼繫結，在管理介面手動繫結
        channel.queueBind(queueName, exchangeName, routingKey);
        channel.queueBind(queueName1, exchangeName, routingKey);

        //5. 建立消費者並接收訊息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                String message = new String(body, "UTF-8");
                System.out.println("[x] Received '" + message + "'");

                channel.basicAck(envelope.getDeliveryTag(), true);
            }
        };

        Consumer consumer1 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body);
                System.out.println("[Y] Receives '" + message + "'");
                channel.basicAck(envelope.getDeliveryTag(), true);
            }
        };

        //6. 設定 Channel 消費者繫結佇列
        channel.basicConsume(queueName, false, consumer);
        channel.basicConsume(queueName, false, consumer1);
    }
}

從如下結果中我們可以看到兩個消費者依次消費三條訊息。

[x] Received 'this is qos msg : 0'
[x] Received 'this is qos msg : 1'
[x] Received 'this is qos msg : 2'
[Y] Receives 'this is qos msg : 3'
[Y] Receives 'this is qos msg : 4'
[Y] Receives 'this is qos msg : 5'
[x] Received 'this is qos msg : 6'
[x] Received 'this is qos msg : 7'
[x] Received 'this is qos msg : 8'
[Y] Receives 'this is qos msg : 9'

當我們將void basicQos(int prefetchSize, int prefetchCount, boolean global)中的 global 設定為 true的時候我們發現並沒有了限流的作用。

[x] Received 'this is qos msg : 0'
[x] Received 'this is qos msg : 1'
[x] Received 'this is qos msg : 2'
[x] Received 'this is qos msg : 3'
[Y] Receives 'this is qos msg : 4'
[x] Received 'this is qos msg : 5'
[Y] Receives 'this is qos msg : 6'
[x] Received 'this is qos msg : 7'
[Y] Receives 'this is qos msg : 8'
[x] Received 'this is qos msg : 9'

TTL

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

這與 Redis 中的過期時間概念類似。我們應該合理使用 TTL 技術，可以有效的處理過期垃圾訊息，從而降低伺服器的負載，最大化的發揮伺服器的效能。

RabbitMQ allows you to set TTL (time to live) for both messages and queues. This can be done using optional queue arguments or policies (the latter option is recommended). Message TTL can be enforced for a single queue, a group of queues or applied for individual messages.

RabbitMQ允許您為訊息和佇列設定TTL（生存時間）。 這可以使用可選的佇列引數或策略來完成（建議使用後一個選項）。 可以對單個佇列，一組佇列強制執行訊息TTL，也可以為單個訊息應用訊息TTL。

​ ——摘自 RabbitMQ 官方文件

1.訊息的 TTL

我們在生產端傳送訊息的時候可以在 properties 中指定 expiration屬性來對訊息過期時間進行設定，單位為毫秒(ms)。

     /**
         * deliverMode 設定為 2 的時候代表持久化訊息
         * expiration 意思是設定訊息的有效期，超過10秒沒有被消費者接收後會被自動刪除
         * headers 自定義的一些屬性
         * */
        //5. 傳送
        Map<String, Object> headers = new HashMap<String, Object>();
        headers.put("myhead1", "111");
        headers.put("myhead2", "222");

        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                .deliveryMode(2)
                .contentEncoding("UTF-8")
                .expiration("100000")
                .headers(headers)
                .build();
        String msg = "test message";
        channel.basicPublish("", queueName, properties, msg.getBytes());

我們也可以後臺管理頁面中進入 Exchange 傳送訊息指定expiration

2.佇列的 TTL

我們也可以在後臺管理介面中新增一個 queue，建立時可以設定 ttl，對於佇列中超過該時間的訊息將會被移除。

死信佇列

死信佇列：沒有被及時消費的訊息存放的佇列

訊息沒有被及時消費的原因：

• a.訊息被拒絕（basic.reject/ basic.nack）並且不再重新投遞 requeue=false

• b.TTL(time-to-live) 訊息超時未消費

• c.達到最大佇列長度

實現死信佇列步驟

• 首先需要設定死信佇列的 exchange 和 queue，然後進行繫結:

  Exchange: dlx.exchange
  Queue: dlx.queue
  RoutingKey: # 代表接收所有路由 key

• 然後我們進行正常宣告交換機、佇列、繫結，只不過我們需要在普通佇列加上一個引數即可: arguments.put("x-dead-letter-exchange",' dlx.exchange' )

• 這樣訊息在過期、requeue失敗、 佇列在達到最大長度時，訊息就可以直接路由到死信佇列!

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class DlxProducer {
    public static void main(String[] args) throws Exception {
                //設定連線以及建立 channel 湖綠
        String exchangeName = "test_dlx_exchange";
        String routingKey = "item.update";
      
        String msg = "this is dlx msg";

        //我們設定訊息過期時間，10秒後再消費 讓訊息進入死信佇列
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                .deliveryMode(2)
                .expiration("10000")
                .build();

        channel.basicPublish(exchangeName, routingKey, true, properties, msg.getBytes());
        System.out.println("Send message : " + msg);

        channel.close();
        connection.close();
    }
}

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class DlxConsumer {
    public static void main(String[] args) throws Exception {
                //建立連線、建立channel忽略 內容可以在上面程式碼中獲取
        String exchangeName = "test_dlx_exchange";
        String queueName = "test_dlx_queue";
        String routingKey = "item.#";

        //必須設定引數到 arguments 中
        Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-dead-letter-exchange", "dlx.exchange");

        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        //將 arguments 放入佇列的宣告中
        channel.queueDeclare(queueName, true, false, false, arguments);

        //一般不用程式碼繫結，在管理介面手動繫結
        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", "#");

        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {

                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");

            }
        };

        //6. 設定 Channel 消費者繫結佇列
        channel.basicConsume(queueName, true, consumer);
    }
}

總結

DLX也是一個正常的 Exchange，和一般的 Exchange 沒有區別，它能在任何的佇列上被指定，實際上就是設定某個佇列的屬性。當這個佇列中有死信時，RabbitMQ 就會自動的將這個訊息重新發布到設定的 Exchange 上去，進而被路由到另一個佇列。可以監聽這個佇列中訊息做相應的處理。