插插插综合,亚洲色视频在线

實(shí)際使用RabbitMQ過程中，如果完全不配置QoS，這樣Rabbit會(huì)盡可能快速地
發(fā)送隊(duì)列中的所有消息到client端。因?yàn)閏onsumer在本地緩存所有的message，
從而極有可能導(dǎo)致OOM或者導(dǎo)致服務(wù)器內(nèi)存不足影響其它進(jìn)程的正常運(yùn)行。所以我們
需要通過設(shè)置Qos的prefetch count來控制consumer的流量。同時(shí)設(shè)置得當(dāng)也會(huì)提高consumer的吞吐量。

prefetch與消息投遞

prefetch允許為每個(gè)consumer指定最大的unacked messages數(shù)目。簡(jiǎn)單來說就是用來指定一個(gè)consumer一次可以從Rabbit中獲取多少條message并緩存在client中(RabbitMQ提供的各種語言的client library)。一旦緩沖區(qū)滿了，Rabbit將會(huì)停止投遞新的message到該consumer中直到它發(fā)出ack。

假設(shè)prefetch值設(shè)為10，共有兩個(gè)consumer。意味著每個(gè)consumer每次會(huì)從queue中預(yù)抓取 10 條消息到本地緩存著等待消費(fèi)。同時(shí)該channel的unacked數(shù)變?yōu)?0。而Rabbit投遞的順序是，先為consumer1投遞滿10個(gè)message，再往consumer2投遞10個(gè)message。如果這時(shí)有新message需要投遞，先判斷channel的unacked數(shù)是否等于20，如果是則不會(huì)將消息投遞到consumer中，message繼續(xù)呆在queue中。之后其中consumer對(duì)一條消息進(jìn)行ack，unacked此時(shí)等于19，Rabbit就判斷哪個(gè)consumer的unacked少于10，就投遞到哪個(gè)consumer中。

總的來說，consumer負(fù)責(zé)不斷處理消息，不斷ack，然后只要unacked數(shù)少于prefetch * consumer數(shù)目，broker就不斷將消息投遞過去。

如何設(shè)置

官方提供的java client可以通過channel來設(shè)置：

channel = connection.createChannel();
channel.basicQos(prefetch);

spring-amqp的話可通過配置文件來配置

<rabbit:listener-container connection-factory="connectionFactory" concurrency="2" prefetch="3">
    <rabbit:listener ref="listener" queue-names="remoting.queue" />
</rabbit:listener-container>

這里需要注意的是，spring-amqp中的prefetch默認(rèn)值是250。

1.0 的spring-amqp默認(rèn)值是1； 2.0版本的為250。具體可看官方文檔
https://docs.spring.io/spring-amqp/docs/2.2.9.RELEASE/reference/html/#async-consumer

客戶端源碼剖析

官方Java客戶端提供了DefaultConsumer和QueueingConsumer兩種類來從queue中獲取消息。其中QueueingConsumer內(nèi)部維護(hù)了一個(gè)阻塞隊(duì)列BlockingQueue，此隊(duì)列就是用來緩存從queue獲取的message。當(dāng)調(diào)用 channel.basicConsume后，broker就會(huì)不斷往consumer投遞message，直到prefetch條。

初始化的時(shí)候，如果不指定BlockingQueue的長(zhǎng)度，默認(rèn)值會(huì)設(shè)為Integer.MAX_VALUE，所以這就解釋了文章開頭所說的如果不設(shè)置Qos的話為什么會(huì)有可能導(dǎo)致OOM，因?yàn)榇藭r(shí)BlockingQueue會(huì)不斷膨脹，消耗內(nèi)存。所以設(shè)置了prefetch后，建議BlockingQueue的長(zhǎng)度(capacity)也初始化為prefetch。

另外需要注意的是，在調(diào)用channel.basicConsume之后，consumer是通過異步方式來抓取message的，通過debug可以發(fā)現(xiàn)BlockingQueue的size是在異步地不斷增長(zhǎng)直到prefetch。而客戶端代碼可以通過consumer.nextDelivery()或consumer.nextDelivery(long timeout)方法來獲取message，其對(duì)應(yīng)的就是BlockingQueue的take()和poll(long timeout)方法。

再來看看spring-amqp的comsumer，大致也一樣。核心類BlockingQueueConsumer

public class BlockingQueueConsumer {
    
    private final BlockingQueue<Delivery> queue;

    //some code
    ...

public BlockingQueueConsumer(ConnectionFactory connectionFactory,
            MessagePropertiesConverter messagePropertiesConverter,
            ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode,
            boolean transactional, int prefetchCount, boolean defaultRequeueRejected,
            Map<String, Object> consumerArgs, boolean exclusive, String... queues) {

        //... some code

        this.queue = new LinkedBlockingQueue<Delivery>(prefetchCount);
    }

BlockingQueueConsumer的構(gòu)造函數(shù)清楚說明了每個(gè)消費(fèi)者內(nèi)部的隊(duì)列大小就是prefetch的大小。

吞吐量、延遲

prefetch并不是說設(shè)置得越大越好。過大可能導(dǎo)致consumer處理不過來，一直在本地緩存的BlockingQueue里呆太久，這樣消息在客戶端的延遲就大大增加；而對(duì)于多個(gè)consumer的情況，則會(huì)分配不均勻，導(dǎo)致有些consumer一直在忙，有些則非常空閑。

然而設(shè)置的過小，又會(huì)令到consumer不能充分工作，因?yàn)槲覀兛傁胨?00％的時(shí)間都是處于繁忙狀態(tài)，而這時(shí)可能會(huì)在處理完一條消息后，BlockingQueue為空，因?yàn)樾碌南⑦€未來得及到達(dá)，所以consumer就處于空閑狀態(tài)了。

prefetch應(yīng)該設(shè)置多大，具體可參考這篇文章

Some queuing theory: throughput, latency and bandwidth

里面詳細(xì)論述吞吐量與prefetch之間的關(guān)系。prefetch的設(shè)置與以下幾點(diǎn)有關(guān)：