>[info] 限流機制 **Con - 限流機制** 為何要限流： * 假設我們有這樣的場景 Rabbitmq服務器有上萬條未處理的消息,我們隨便打開一個Con - Client,會造成:巨量的消息瞬間全部推送過來,然而我們單個客戶端無法同時處理這么多數據!此時很有可能導致服務器崩潰，嚴 重的可能導致線上的故障。 * 還有一些其他的場景，比如說單個Pro一分鐘產生了幾百條數據,但是單個Con一分鐘可能只能處理60條,這個時候Pro-Con肯定是不平衡的。通常Pro是沒辦法做限制的。所以Con肯定需要做一些限流措施，否則如果 超出最大負載，可能導致Con性能下降，服務器卡頓甚至崩潰等一系列嚴重后果 RabbitMQ提供了一種qos (服務質量保證)功能,即在非自動確認消息的前提下,如果一定數目的消息 (通過基于Con或者channel設置Qos的值) 未被確認前,不消費新的消息 >[] 不能設置自動簽收功能(autoAck = false) 如果消息未被確認,就不會到達Con,目的就是給Pro減壓 ***** **場景：** * 假設當前生產者已經推送了上萬條信息給MQ，那么消費者一下子需要消費上萬條，負載大。 * 生產者秒推幾百條消息給MQ，消費者只能消費幾十條，超過負載。 ***** **限流設置API：** ``` // 消費者代碼：一次只拿20條并處理 $channel->basic_qos($prefetchSize, 20, $global); ``` * prefetchSize: 單條消息的大小限制，Con通常設置為0，表示不做限制 * prefetchCount: 一次最多能處理多少條消息 * global: 是否將上面設置true應用于channel級別還是取false代表Con級別 >[] prefetchSize和global 這兩項,RabbitMQ沒有實現,暫且不研究 prefetchCount在 <span style='color:red'>autoAck=false</span> 的情況下生效,即在自動應答的情況下該值無效 **手工ACK：（應答MQ已經處理完成）** ``` $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); ``` 調用這個方法就會主動回送給Broker一個應答，表示這條消息我處理完了，你可以給我下一條了。參數multiple表示是否批量簽收，由于我們是一次處理一條消息，所以設置為false ***** **代碼示例：basic_qos.php** ~~~ <?php include(__DIR__ . '/config.php'); use PhpAmqpLib\Connection\AMQPStreamConnection; $connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST); $channel = $connection->channel(); $channel->queue_declare('qos_queue', false, true, false, false); //第二個參數代表：每次只消費100條 $channel->basic_qos(null, 109, null); function process_message($message) { /*業務邏輯*/ //消費完消息之后進行應答，告訴rabbit我已經消費了，可以發送下一組了 $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); } $channel->basic_consume('qos_queue', '', false, false, false, false, 'process_message'); while ($channel->is_consuming()) { // After 10 seconds there will be a timeout exception. $channel->wait(null, false, 30000); } ~~~ 1. 運行代碼生成隊列  2. 推送5條消息  3. 消費  * 我們之前是注釋掉手工ACK方法，然后啟動消費端和生產端，當時Con只消費了1條消息,這是因為我們設置了手工簽收,并且設置了一次只處理一條消息,當我們沒有回送ACK應答時，Broker端就認為Con還沒有處理完 這條消息,基于這種限流機制就不會給Con發送新的消息了,所以Con那時只打印了一條消息