RabbitMQ——消費端限流、TTL、死信隊列

消費端限流#

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

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

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

2.限流的 api 講解#

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

<code>Copy/**
* 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;
/<code>

• 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);

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

<code>Copyimport 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();
    }
}
/<code>

這裡我們創建一個消費者，通過以下代碼來驗證限流效果以及 global 參數設置為 true 時不起作用.。我們通過Thread.sleep(5000); 來讓 ack 即處理消息的過程慢一些，這樣我們就可以從後臺管理工具中清晰觀察到限流情況。

<code>Copyimport 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 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);

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


                channel.basicAck(envelope.getDeliveryTag(), true);
            }
        };
        //6. 設置 Channel 消費者綁定隊列
        channel.basicConsume(queueName, false, consumer);
        channel.basicConsume(queueName, false, consumer1);
    }
}
/<code>

我們從下圖中發現 Unacked值一直都是 3 ，每過 5 秒 消費一條消息即 Ready 和 Total 都減少 3，而 Unacked的值在這裡代表消費者正在處理的消息，通過我們的實驗發現了消費者一次性最多處理 3 條消息，達到了消費者限流的預期功能。

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

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)。

<code>Copy\t /**
         * deliverMode 設置為 2 的時候代表持久化消息
         * expiration 意思是設置消息的有效期，超過10秒沒有被消費者接收後會被自動刪除
         * headers 自定義的一些屬性
         * */
        //5. 發送
        Map<string> headers = new HashMap<string>();
        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());
/<string>/<string>/<code>

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

2.隊列的 TTL#

我們也可以在後臺管理界面中新增一個 queue，創建時可以設置 ttl，對於隊列中超過該時間的消息將會被移除。

死信隊列#

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

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

• a.消息被拒絕（basic.reject/ basic.nack）並且不再重新投遞 requeue=false
• b.TTL(time-to-live) 消息超時未消費
• c.達到最大隊列長度

實現死信隊列步驟#

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

<code>Copyimport 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 {
\t\t\t\t//設置連接以及創建 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();
    }
}

/<code>

<code>Copyimport 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 {
\t\t\t\t//創建連接、創建channel忽略 內容可以在上面代碼中獲取
        String exchangeName = "test_dlx_exchange";
        String queueName = "test_dlx_queue";
        String routingKey = "item.#";

        //必須設置參數到 arguments 中
        Map<string> arguments = new HashMap<string>();
        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);
    }
}

/<string>/<string>/<code>

總結#

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

出處：https://www.cnblogs.com/haixiang/p/10905189.html

本站使用「CC BY 4.0」創作共享協議，轉載請在文章明顯位置註明作者及出處。

閱讀更多 咔咔侃技術 的文章

關鍵字: 死信 限流 FLOW