prefetch_count參數的含義
先從AMQP（Advanced Message Queuing Protocol，及高級消息隊列協議，RabbitMQ實現了此協議的0-9-1版本的大部分內容）和RabbitMQ的具體實現去理解prefetch_count參數的含義，可以查閱對應的文檔（見文末參考資料）。AMQP 0-9-1定義了basic.qos方法去限制消費者基於某一個Channel或者Connection上未進行ack的最大消息數量上限。basic.qos方法支持兩個參數：
- global：布爾值。
- prefetch_count：整數。
這兩個參數在AMQP 0-9-1定義中的含義和RabbitMQ具體實現時有所不同，見下表：
或者用簡潔的英文表格理解：
這里畫一個圖理解一下：
上圖僅僅為了區分協議本身和RabbitMQ中實現的不同，接着說說prefetch_count對於消費者（線程）和待消費消息的作用。假定一個前提：RabbitMQ客戶端從RabbitMQ服務端獲取到隊列消息的速度比消費者線程消費速度快，目前有兩個消費者線程共用一個Channel實例。當global參數為false時候，效果如下：
而當global參數為true時候，效果如下：
在消費者線程處理速度遠低於RabbitMQ客戶端從RabbitMQ服務端獲取到隊列消息的速度的場景下，prefetch_count條未進行ack的消息會暫時存放在一個隊列（准確來說是阻塞隊列，然后阻塞隊列中的消息任務會流轉到一個列表中遍歷回調消費者句柄，見下一節的源碼分析）中等待被消費者處理。這部分消息會占據JVM的堆內存，所以在性能調優或者設定應用程序的初始化和最大堆內存的時候，如果剛好用到RabbitMQ的消費者，必須要考慮這些"預取消息"的內存占用量。不過值得注意的是：prefetch_count是RabbitMQ服務端的參數，它的設置值或者快照都不會存放在RabbitMQ客戶端。同時需要注意prefetch_count生效的條件和特性（從參數設置的一些demo和源碼上感知）：
- prefetch_count參數僅僅在 basic.consume的 autoAck參數設置為 false的前提下才生效，也就是不能使用自動確認，自動確認的消息沒有辦法限流。
- basic.consume如果在非自動確認模式下忘記了手動調用 basic.ack，那么 prefetch_count正是未 ack消息數量的最大上限。
- prefetch_count是由 RabbitMQ服務端控制，一般情況下能保證各個消費者線程中的未 ack消息分發是均衡的，這點筆者猜測是 consumerTag起到了關鍵作用。
RabbitMQ客戶端中prefetch_count源碼跟蹤
編寫本文的時候引入的RabbitMQ客戶端版本為：
com.rabbitmq:amqp-client:5.9.0
上面說了這么多都只是根據官方的文檔或者博客中的理論依據進行分析，其實更加根本的分析方法是直接閱讀RabbitMQ的Java客戶端源碼，主要是針對basic.qos和basic.consume兩個方法，對應的是
com.rabbitmq.client.impl.ChannelN#basicQos()和
com.rabbitmq.client.impl.ChannelN#basicConsume()兩個方法。先看ChannelN#basicQos()：
這里的basicQos()方法多了一個prefetchSize參數，用於限制分發內容的大小上限，默認值0代表無限制，而prefetchCount的取值范圍是[0,65535]，取值為0也是代表無限制。這里的ChannelN#basicQos()實現中直接封裝basic.qos方法參數進行一次RPC調用，意味着直接更變RabbitMQ服務端的配置，即時生效，同時參數值完全沒有保存在客戶端代碼中，印證了前面一節的結論。接着看ChannelN#basicConsume()方法：
上圖已經把關鍵部分用紅圈圈出，因為整個消息消費過程是異步的，涉及太多的類和方法，這里不全量貼出，整理了一個流程圖：
整個消息消費過程，prefetch_count參數並未出現在客戶端代碼中，又再次印證了前面一節的結論，即prefetch_count參數的行為和作用完全由RabbitMQ服務端控制。而最終Customer或者常用的DefaultCustomer句柄是在WorkPoolRunnable中回調的，這類任務的執行線程來自於ConsumerWorkService內部的線程池，而這個線程池又使用了
Executors.newFixedThreadPool()去構建，使用了默認的線程工廠類，因此在Customer#handleDelivery()方法內部打印的線程名稱的樣子是pool-1-thread-*。
這里
VariableLinkedBlockingQueue就是前一節中的message queue的原型
prefetch_count參數使用
設置prefetch_count參數比較簡單，就是調用Channel#basicQos()方法：
public class RabbitQos { static String QUEUE = "qos.test"; public static void main(String[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("localhost"); connectionFactory.setPort(5672); connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE, true, false, false, null); channel.basicQos(2); channel.basicConsume("qos.test", false, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("1------" + Thread.currentThread().getName()); sleep(); } }); channel.basicConsume("qos.test", false, new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("2------" + Thread.currentThread().getName()); sleep(); } }); for (int i = 0; i < 20; i++) { channel.basicPublish("", QUEUE, MessageProperties.TEXT_PLAIN, String.valueOf(i).getBytes()); } sleep(); } private static void sleep() { try { Thread.sleep(Long.MAX_VALUE); } catch (Exception ignore) { } } }上面是原生的amqp-client的寫法，如果使用了spring-amqp（spring-boot-starter-amqp），可以通過配置文件中的
spring.rabbitmq.listener.direct.prefetch屬性指定所有消費者線程的prefetch_count，如果要針對部分消費者線程進行該屬性的設置，則需要針對
RabbitListenerContainerFactory進行改造。
prefetch_count參數最佳實踐
關於prefetch_count參數的設置，RabbitMQ官方有一篇文章進行了分析：《Finding bottlenecks with RabbitMQ 3.3》。該文章分析了消息流控的整個流程，其中提到了prefetch_count參數的一些指標：
這里指出了，如果prefetch_count的值超過了30，那么網絡帶寬限制開始占主導地位，此時進一步增加prefetch_count的值就會變得收效甚微。也就是說，「官方是建議把prefetch_count設置為30」。這里再參看一下spring-boot-starter-amqp中對此參數定義的默認值，具體是
AbstractMessageListenerContainer中的DEFAULT_PREFETCH_COUNT：
如果沒有通過
spring.rabbitmq.listener.direct.prefetch進行覆蓋，那么使用spring-boot-starter-amqp中的注解定義的消費者線程中設置的prefetch_count就是250。
筆者認為，應該綜合帶寬、每條消息的數據報大小、消費者線程處理的速率等等角度去考慮prefetch_count的設置。總結如下（個人經驗僅供參考）：
- 當消費者線程的處理速度十分慢，而隊列的消息量十分少的場景下，可以考慮把 prefetch_count設置為 1。
- 當隊列中的每條消息的數據報十分大的時候，要計算好客戶端可以容納的未 ack總消息量的內存極限，從而設計一個合理的 prefetch_count值。
- 當消費者線程的處理速度十分快，遠遠大於 RabbitMQ服務端的消息分發，在網絡帶寬充足的前提下，設置可以把 prefetch_count值設置為 0，不做任何的消息流控。
- 一般場景下，建議使用 RabbitMQ官方的建議值 30或者 spring-boot-starter-amqp中的默認值 250。
小結
小結一下：
- prefetch_count是 RabbitMQ服務端的參數，設置后即時生效。
- prefetch_count對於 AMQP-0-9-1中的定義與 RabbitMQ中的實現不完全相同。
- prefetch_count值設置建議使用框架提供的默認值或者通過分組實驗結合數據報大小進行計算和評估出一個合理值。