RabbitMQ使用技巧 · 個人筆記

1.RabbitMQ之延時隊列 ``` 在消息生產者處 use PhpAmqpLib\Wire\AMQPTable; （1）通過隊列屬性x-message-ttl設置，投遞到該隊列中的所有消息都有相同的過期時間第一種寫法： $tale = new AMQPTable([ 'x-dead-letter-exchange' => $delayExName, 'x-message-ttl' => $ttl, //消息存活時間，單位毫秒 'x-dead-letter-routing-key' => $queueName ]); 第二種寫法： $tale = new AMQPTable(); $tale->set('x-dead-letter-exchange', 'delay_exchange'); $tale->set('x-dead-letter-routing-key','delay_exchange'); $tale->set('x-message-ttl',10000); //10秒放在隊列聲明第7個參數 $channel->queue_declare('cache_queue',false,true,false,false,false,$tale); （2）通過消息屬性expiration設置消息本身的有效期 $msg = new AMQPMessage('Hello World'.'3000',array( 'expiration' => intval(18000), 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT )); （3）注意事項 * 如果兩種同時使用，會以兩者之間較小的那個數值為準，單位都為毫秒 * 通過隊列設置的，一旦消息過期，就會從隊列中抹去，因為過期的消息肯定在隊列頭部，RabbitMQ只需要定期處理頭部過期消息即可。 * 而單獨設置有效期的，如果要刪除則需要遍歷整個隊列，所以采取消費時判定是否過期處理刪除（4）隊列有效期 $tale = new \PhpAmqpLib\\Wire\AMQPTable(); $tale->set('x-expires', 60000 ); 通過channel.queueDeclare方法中到x-expires參數是可以控制隊列在指定時間未被使用過后刪除，未被使用包括以下三點： * 沒有任何消費者 * 未被重新聲明過期時間 * 未調用過Basic.Get命令 ``` 2.工作隊列 ``` （1）同時運行兩個`worker.php`腳本。它們都會從隊列中獲取消息 # shell 1 php worker.php # => [*] Waiting for messages. To exit press CTRL+C # => [x] Received 'First message.' # => [x] Received 'Third message...' # => [x] Received 'Fifth message.....' # shell 2 php worker.php # => [*] Waiting for messages. To exit press CTRL+C # => [x] Received 'Second message..' # => [x] Received 'Fourth message....' ``` （2）在默認情況下，RabbitMQ 將按順序將隊列中的消息分發給消費者。平均每個消費者獲得相同數量的消息。這種奮發消息的方式叫做循環（3）公平調度消息消費者代碼 $channel->basic_qos(null, 1, null); 我們可以通過設置 basic_qos 第二個參數 prefetch_count = 1。這一項告訴RabbitMQ不要一次給一個消費者發送多個消息。或者換一種說法，在確認前一個消息之前，不要向消費者發送新的消息。相反，新的消息將發送到一個處于空閑的消費者又或者只有consumer已經處理并確認了上一條message時queue才分派新的message給它 3.消息確認機制 (ack) 消息消費者代碼在默認情況下，消息確認機制是關閉的。現在是時候開啟消息確認機制，將basic_consumer的第四個參數設置為false(true表示不開啟消息確認)，并且工作進程處理完消息后發送確認消息。 ``` # 處理消息回調函數 $callback = function($msg){ echo " [x] Received ", $msg->body, "\n"; sleep(substr_count($msg->body, '.')); echo " [x] Done", "\n"; $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']); }; # 開啟消息確認 $channel->basic_consume('task_queue', '', false, false, false, false, $callback); ``` 4.消息持久化（1）隊列聲明為持久化：要將 queue_declare 的第三個參數durable設置為 true ： $channel->queue_declare($queue, false, true, false, false); RabbitMQ不允許使用不同的參數重新定義一個已經存在的隊列，并且會返回一個錯誤給任何嘗試做此事的程序（2）消息持久化： $msg = new AMQPMessage($data, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT) ); 或者'delivery_mode' =>2 （3）交換機持久化:消息生產者代碼 $channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete); 設置第四個參數$durable為true,即開啟交換機持久化交換機的持久化其實就是相當于將交換機的屬性在服務器內部保存，當MQ的服務器發生意外或關閉之后，重啟RabbitMQ時不需要重新手動或執行代碼去建立交換機，交換機會自動建立，相當于一直存在。 5.消息主題(topics) 強烈注意：隊列可以綁定多個路由鍵到交換機上發送到 topic exchange 的消息不能任意命名一個 routing key - 它必須是由一個.劃分單詞列表。這些單詞可以是任意的，但它們通常指定與消息相關聯的一些功能。這里有幾個有效的 routing key: "stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit" , routing key 中可以包含多個單詞，最多可以達到255個字節。 binding key也必須是相同的形式。topic exchange背后的邏輯類似于 direct exchange - 使用特定routing key1 發送的消息將被傳遞到與binding key綁定的所有隊列。但是，binding key 有兩個重要的特殊情況： *(星號) 可以代替一個單詞 #(哈希) 可以匹配0個或多個單詞我們將發送所有描述動物的消息。消息將使用由三個單詞（兩個點）組成的routing key發送。其中第一個單詞描述速度，第二個描述顏色，第三個描述種類：”..”。我們接著創建三個綁定：Q1 binding key “*.orange.*”, Q2 binding key “*.*.rabbit” “lazy.#”。這三個綁定可以解釋為： Q1 對所有橙色的動物感興趣。 Q2 想要獲取有關兔子的一切消息，以及所有惰性動物的一切。一條 routing key 為 “quick.orange.rabbit” 的消息將傳遞上面的到兩個對列。routing key 為 “lazy.orange.elephant” 的消息也將傳遞上面的到兩個對列。另外 “lazy.pink.rabbit” 消息將只會被傳遞到Q2一次，即使它匹配了兩個 binding key。”quick.brown.fox” 不匹配任何 binding key, 所以它將被丟棄。如果我們不遵守以上的規則發送 routing key 為一個或者四個單詞的消息會發生什么？比如，”orange” 或者 “quick.orange.male.rabbit”。那么我們將丟失這些消息，因為它們不匹配任何 binding key。另一方面，”lazy.orange.male.rabbit” 即使它有四個單詞，但它能匹配最后一個綁定，并且將被傳遞到Q2中。