RabbitMQ高級應用之消費端限流策略basicQos詳解

更新時間：2023年08月28日 10:08:26 作者：瘋狂的帆

這篇文章主要介紹了RabbitMQ高級應用之消費端限流策略basicQos詳解,高并發(fā)情況下,隊列里面一瞬間就就積累了上萬條數(shù)據(jù),但是消費者無法同時處理這么多請求,這種場景下我們就需要對消費端進行限流,需要的朋友可以參考下

業(yè)務場景

高并發(fā)情況下，隊列里面一瞬間就就積累了上萬條數(shù)據(jù)，但是消費者無法同時處理這么多請求，這個時候當我們打開客戶端，瞬間就有巨量的信息給推送過來

但是客戶端是沒有辦法同時處理這么多數(shù)據(jù)的，結果就是消費者（客戶端）掛掉了…

這種場景下我們就需要對消費端進行限流

限流策略實現(xiàn)

限流策略關鍵代碼：

channel.basicQos();

編寫生產(chǎn)者

// 生產(chǎn)者
public class Producer {
    private static final String QUEUE_NAME = "queue_limit_1";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        for (int i = 0; i < 100; i++) {
            channel.basicPublish("", QUEUE_NAME, null, ("消費端限流策略—測試數(shù)據(jù)：" + i).getBytes());
        }
        channel.close();
        connection.close();
    }
}

編寫消費者1

// 消費者1
public class Consumer {
    private static final String QUEUE_NAME = "queue_limit_1";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者1接收到信息：" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    }
}

編寫消費者2

// 消費者2
public class Consumer2 {
    private static final String QUEUE_NAME = "queue_limit_1";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        /**  設置限流機制
         *  param1: prefetchSize，消息本身的大小 如果設置為0  那么表示對消息本身的大小不限制
         *  param2: prefetchCount，告訴rabbitmq不要一次性給消費者推送大于N個消息
         *  param3：global，是否將上面的設置應用于整個通道，false表示只應用于當前消費者
         */
        channel.basicQos(0, 5, false);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消費者2接收到信息：" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    }
}

運行結果

在這里插入圖片描述

小結

限流的核心代碼就是 channel.basicQos();
限流情況 ack 不能設置自動簽收，一定要手動簽收 channel.basicQos()

/**
     * @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
     */
    void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

該方法的作用是：進行消費端的限流