private boolean receiveAndExecute(final BlockingQueueConsumer consumer) throws Exception { // NOSONAR
    PlatformTransactionManager transactionManager = getTransactionManager();
    if (transactionManager != null) {...事務操作，不關(guān)注
    }
    //接受消息并進行處理
    return doReceiveAndExecute(consumer);
}

若是執(zhí)行nextMessage()沒有獲取到消息，那么執(zhí)行break操作，最終會導致上面的receiveAndExecute()方法返回false。而receiveAndExecute()的值可以決定是否動態(tài)擴容

private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Exception { //NOSONAR
    Channel channel = consumer.getChannel();
    //默認txSize=1
    for (int i = 0; i < this.txSize; i++) {

        //在內(nèi)存隊列中獲取消息
        Message message = consumer.nextMessage(this.receiveTimeout);
        //未獲取到消息，開始下次循環(huán)
        if (message == null) {
            break;
        }
        try {
            //執(zhí)行業(yè)務邏輯  
            executeListener(channel, message);
        } 
       ...catch操作，不關(guān)注
    }
    //沒有獲取到消息，這個方法會返回false
    return consumer.commitIfNecessary(isChannelLocallyTransacted());
}

public boolean commitIfNecessary(boolean locallyTransacted) throws IOException {
    //此處直接返回false
    if (this.deliveryTags.isEmpty()) {
        return false;
    }
    ...
    return true;
}

在內(nèi)存中獲取消息

由于配置了setPrefetchCount參數(shù)，所以內(nèi)存會去MQ中預取配置的消息數(shù)，放到本地的BlockingQueue中。
配置詳見：
【RabbitMQ-2】RabbitMQ的并發(fā)參數(shù)（concurrency和prefetch）

未獲取到消息

public Message nextMessage(long timeout) throws InterruptedException,
ShutdownSignalException {
    if (logger.isTraceEnabled()) {
        logger.trace("Retrieving delivery for " + this);
    }
    checkShutdown();
    if (this.missingQueues.size() > 0) {
        checkMissingQueues();
    }
    //poll的API描述：檢索并刪除此隊列的頭，等待指定的等待時間（如有必要）使元素變?yōu)榭捎谩?    Message message = handle(this.queue.poll(timeout, TimeUnit.MILLISECONDS));
    //cancelled默認false不會執(zhí)行改邏輯
    if (message == null && this.cancelled.get()) {
        throw new ConsumerCancelledException();
    }
    //未獲取到消息返回null。
    return message;
}

2.2 checkAdjust()—動態(tài)擴容業(yè)務

由上面源碼可以若是沒有獲取到消息，receivedOk返回false（注：若是獲取到消息，但是NACK，receivedOk返回值依舊是true）。

如何保證是連續(xù)獲取或者連續(xù)空轉(zhuǎn)的？
答案：因為mainloop()一直循環(huán)，每次均在本地queue獲取消息（最長阻塞1s）。若連續(xù)9次均未獲取到消息，第10次獲取到消息，那么會重置consecutiveIdles=0。

private void checkAdjust(boolean receivedOk) {
    //成功獲取到消息
    if (receivedOk) {
        if (isActive(this.consumer)) {
            //連續(xù)空轉(zhuǎn)標識設置為0
            this.consecutiveIdles = 0;
            //consecutiveActiveTrigger默認為10
            if (this.consecutiveMessages++>SimpleMessageListenerContainer.this.consecutiveActiveTrigger) {
               //開啟一個消費者線程
                considerAddingAConsumer();
                //練習消費的標識設置為0
                this.consecutiveMessages = 0;
            }
        }
    } else {
        this.consecutiveMessages = 0;
        if (this.consecutiveIdles++>SimpleMessageListenerContainer.this.consecutiveIdleTrigger) {
            considerStoppingAConsumer(this.consumer);
            this.consecutiveIdles = 0;
        }
    }
}

開啟一個消費者線程：

private void considerAddingAConsumer() {
    //加鎖
    synchronized(this.consumersMonitor) {
      //若是當前consumers數(shù)量小于配置maxConcurrentConsumers
        if (this.consumers != null && this.maxConcurrentConsumers != null && this.consumers.size() < this.maxConcurrentConsumers) {
            long now = System.currentTimeMillis();
            //開啟消費者有間隔時間
            if (this.lastConsumerStarted + this.startConsumerMinInterval < now) {
                //增加消費者。
                this.addAndStartConsumers(1);
                this.lastConsumerStarted = now;
            }
        }
    }
}

開啟消費者的操作

protected void addAndStartConsumers(int delta) {
    synchronized(this.consumersMonitor) {
        if (this.consumers != null) {
            //每一次循環(huán)均是創(chuàng)建一個消費者
            for (int i = 0; i < delta; i++) {
                 //判斷是否創(chuàng)建消費者
                if (this.maxConcurrentConsumers != null && this.consumers.size() >= this.maxConcurrentConsumers) {
                    break;
                }
                //創(chuàng)建消費者
                BlockingQueueConsumer consumer = createBlockingQueueConsumer();
                //(核心)屬性的consumers+1
                this.consumers.add(consumer);
                AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer);
                if (logger.isDebugEnabled()) {
                    logger.debug("Starting a new consumer: " + consumer);
                }  
                //使用內(nèi)部的線程池執(zhí)行
                getTaskExecutor().execute(processor);
                //發(fā)布創(chuàng)建消費者事件
                if (this.getApplicationEventPublisher() != null) {
                    this.getApplicationEventPublisher().publishEvent(new AsyncConsumerStartedEvent(this, consumer));
                }
                try {
                    //線程執(zhí)行完run方法后，線程中的CountDownLatch-1。
                    //若線程池沒有資源，那么會在此處阻塞（默認60s）
                    //阻塞完畢，startupException返回null。即成功創(chuàng)建
                    FatalListenerStartupException startupException = processor.getStartupException();
                    //若是線程池資源不足，只是返回null，不會執(zhí)行下面分支。
                    if (startupException != null) {
                        this.consumers.remove(consumer);
                        throw new AmqpIllegalStateException("Fatal exception on listener startup", startupException);
                    }
                } catch(InterruptedException ie) {
                    Thread.currentThread().interrupt();
                } catch(Exception e) {
                    consumer.stop();
                    logger.error("Error starting new consumer", e);
                    this.cancellationLock.release(consumer);
                    this.consumers.remove(consumer);
                }
            }
        }
    }
}

1. 上述代碼中，創(chuàng)建消費者線程是同步的流程，即某個消費者線程加鎖去創(chuàng)建。某創(chuàng)建時線程池沒有資源，會阻塞消費者線程。
2. 若線程池沒有資源，阻塞完畢后，只是打印異常日志，并拋出異常，此時內(nèi)存中消費者個數(shù)為n+1個，但是只有n個線程可以消費消息
3. 當連續(xù)10次空轉(zhuǎn)時consecutiveIdles =10，且消費者線程n+1，會回收臨時擴展的消費者線程。

private void considerStoppingAConsumer(BlockingQueueConsumer consumer) {
    synchronized(this.consumersMonitor) {
        if (this.consumers != null && this.consumers.size() > this.concurrentConsumers) {
            long now = System.currentTimeMillis();
            if (this.lastConsumerStopped + this.stopConsumerMinInterval < now) {
                //回收消費者的核心方式
                consumer.basicCancel(true);
                //本地消費者集合移除消費者
                this.consumers.remove(consumer);
                if (logger.isDebugEnabled()) {
                    logger.debug("Idle consumer terminating: " + consumer);
                }
                this.lastConsumerStopped = now;
            }
        }
    }
}

上面說到，內(nèi)存中的消費者數(shù)量n+1，但是有效的消費者n個。當回收消費者時會回收有效的消費者使得內(nèi)存消費者數(shù)量n個，有效消費者數(shù)量n-1個。

若是線程池資源不足，且配置了消費者動態(tài)擴展參數(shù)后，最終會導致有效的消費者數(shù)量為0，導致消息的大量積壓！??！

注：RabbitMQ使用默認的new SimpleAsyncTaskExecutor()開啟消費者線程，即每當使用線程是，均是new出來的。

總結(jié)：不推薦使用自定義配置的線程池，若使用，每次增加隊列時均需要注意配置好線程數(shù)。