前提
在某一次用戶標簽服務中大量用到異步流程，使用了RabbitMQ進行解耦。其中，為了提高消費者的處理效率針對了不同節點任務的消費者線程數和prefetch_count參數都做了調整和測試，得到一個相對合理的組合。這里深入分析一下prefetch_count參數在RabbitMQ中的作用。
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具體實現時有所不同，見下表：
global參數值 |
AMQP 0-9-1中prefetch_count參數的含義 |
RabbitMQ中prefetch_count參數的含義 |
|---|---|---|
false |
prefetch_count值在當前Channel的所有消費者共享 |
prefetch_count對於基於當前Channel創建的消費者生效 |
true |
prefetch_count值在當前Connection的所有消費者共享 |
prefetch_count值在當前Channel的所有消費者共享 |
或者用簡潔的英文表格理解：
global |
prefetch_count in AMQP 0-9-1 |
prefetch_count in RabbitMQ |
|---|---|---|
false |
Per channel limit |
Per customer limit |
true |
Per connection limit |
Per channel limit |
這里畫一個圖理解一下：
上圖僅僅為了區分協議本身和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值設置建議使用框架提供的默認值或者通過分組實驗結合數據報大小進行計算和評估出一個合理值。
彩蛋
筆者把文章發布到公眾號和朋友圈后，筆者的師傅作了點評，指出其中的一點不足：
確實如此，prefetch_count的本質作用就是消費者的流控，官方的那篇文章也提到了網絡和帶寬的重要性，所以要考慮RTT（Round-Trip Time，往返時延），這里的RTT概念來源於《計算機網絡原理》：
The RTT includes packet-propagation delays, packet-queuing delays and packet -processing delay.
也就是說RTT = 數據包傳播時延（往返）+ 數據包排隊時延（路由器和交換機的）+ 數據處理時延（應用程序處理耗時，用在本文的場景就是消費者處理消息的耗時）。假設RTT中只計算網絡的時延，不包含數據處理的時延，那么數據包往返需要2RTT，也就是一條消費消息處理的數據包的往返，RTT越大，那么數據傳輸成本越高，應該允許客戶端"預取"更多的未ack消息避免消費者線程等待。這樣就可以計算出單個消費者線程處理達到最飽和狀態下的"預取"消息量：prefetch_count = 2RTT / 消費者線程處理單條消息的耗時。依照此概念舉例：
- 當
RTT為30ms，而消費者線程處理單條消息的耗時為10ms，此時，消費速率占優勢，可以考慮把prefetch_count設置為6或者更大的值（考慮堆內存極限的限制）。 - 當
RTT為30ms，而消費者線程處理單條消息的耗時為200ms，RTT占優勢，消費速率滯后，此時考慮把prefetch_count設置為1即可。
思考：為什么spring-boot-starter-amqp把prefetch_count默認值設置為250這么高的值，很少開發者改動它卻沒有出現明顯問題？
（本文完 c-4-d e-a-20201017）