快捷導(dǎo)航

RabbitMQ高級(jí)應(yīng)用之消費(fèi)端限流策略basicQos詳解

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

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

業(yè)務(wù)場(chǎng)景

高并發(fā)情況下，隊(duì)列里面一瞬間就就積累了上萬(wàn)條數(shù)據(jù)，但是消費(fèi)者無(wú)法同時(shí)處理這么多請(qǐng)求，這個(gè)時(shí)候當(dāng)我們打開(kāi)客戶端，瞬間就有巨量的信息給推送過(guò)來(lái)

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

這種場(chǎng)景下我們就需要對(duì)消費(fèi)端進(jìn)行限流

限流策略實(shí)現(xiàn)

限流策略關(guā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, ("消費(fèi)端限流策略—測(cè)試數(shù)據(jù)：" + i).getBytes());
        }
        channel.close();
        connection.close();
    }
}

編寫消費(fèi)者1

// 消費(fèi)者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("消費(fèi)者1接收到信息：" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    }
}

編寫消費(fèi)者2

// 消費(fèi)者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);
        /**  設(shè)置限流機(jī)制
         *  param1: prefetchSize，消息本身的大小 如果設(shè)置為0  那么表示對(duì)消息本身的大小不限制
         *  param2: prefetchCount，告訴rabbitmq不要一次性給消費(fèi)者推送大于N個(gè)消息
         *  param3：global，是否將上面的設(shè)置應(yīng)用于整個(gè)通道，false表示只應(yīng)用于當(dāng)前消費(fèi)者
         */
        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("消費(fèi)者2接收到信息：" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    }
}

運(yùn)行結(jié)果

在這里插入圖片描述

小結(jié)

限流的核心代碼就是 channel.basicQos();
限流情況 ack 不能設(shè)置自動(dòng)簽收，一定要手動(dòng)簽收 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;

該方法的作用是：進(jìn)行消費(fèi)端的限流

param1：prefetchSize，消息本身的大小如果設(shè)置為0 那么表示對(duì)消息本身的大小不限制
param2：prefetchCount，告訴rabbitmq不要一次性給消費(fèi)者推送大于N個(gè)消息
param3：global，是否將上面的設(shè)置應(yīng)用于整個(gè)通道
- false：表示只應(yīng)用于當(dāng)前消費(fèi)者
- true：表示當(dāng)前通道的所有消費(fèi)者都應(yīng)用這個(gè)限流策略

到此這篇關(guān)于RabbitMQ高級(jí)應(yīng)用之消費(fèi)端限流策略basicQos詳解的文章就介紹到這了,更多相關(guān)RabbitMQ消費(fèi)端限流策略basicQos內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: