>[info] 重回隊列機制 **ACK & NACK** 當我們設置 ``` autoACK=false ``` 時,就可以使用手工ACK方式了,其實手工方式包括了手工ACK與NACK 當我們手工 ACK 時,會發送給Broker一個應答,代表消息處理成功,Broker就可回送響應給Pro. NACK 則表示消息處理失敗,如果設置了重回隊列,Broker端就 會將沒有成功處理的消息重新發送. ***** **使用方式：** * Con消費時,如果由于業務異常,我們可以手工 NACK 記錄日志,然后進行補償API：void basicNack(long deliveryTag, boolean multiple, boolean requeue)API：void basicAck(long deliveryTag, boolean multiple) * 如果由于服務器宕機等嚴重問題,我們就需要手工 ACK 保障Con消費成功 ***** **重回隊列：** * 重回隊列是為了對沒有處理成功的消息,將消息重新投遞給Broker * 重回隊列,會把消費失敗的消息重新添加到隊列的尾端,供Con繼續消費 * 一般在實際應用中,都會關閉重回隊列,即設置為false ***** **代碼實現：basic_nack.php** ~~~ <?php /** * - Start this consumer in one window by calling: php demo/basic_nack.php * - Then on a separate window publish a message like this: php demo/amqp_publisher.php good * that message should be "ack'ed" * - Then publish a message like this: php demo/amqp_publisher.php bad * that message should be "nack'ed" */ include(__DIR__ . '/config.php'); use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Exchange\AMQPExchangeType; $exchange = 'router'; $queue = 'msgs'; $consumerTag = 'consumer'; $connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST); $channel = $connection->channel(); $channel->queue_declare($queue, false, true, false, false); $channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false); $channel->queue_bind($queue, $exchange); /** * @param \PhpAmqpLib\Message\AMQPMessage $message */ function process_message($message) { /* if(插入成功){ echo "將消息刪除："; 服務器死掉，相當于exit; $message->ack(true); }else if(插入失敗){ echo "將消息不要刪除，等著下次消費"; $message->nack(true); }*/ if ($message->body == 'good') { $message->ack(); } else { echo "成功收到消息，消息內容為：".$message->body ; echo "將消息打回,重回隊列："; $message->nack(true); } // Send a message with the string "quit" to cancel the consumer. if ($message->body === 'quit') { $message->getChannel()->basic_cancel($message->getConsumerTag()); } } $channel->basic_consume($queue, $consumerTag, false, false, false, false, 'process_message'); /** * @param \PhpAmqpLib\Channel\AMQPChannel $channel * @param \PhpAmqpLib\Connection\AbstractConnection $connection */ function shutdown($channel, $connection) { $channel->close(); $connection->close(); } register_shutdown_function('shutdown', $channel, $connection); // Loop as long as the channel has callbacks registered while ($channel->is_consuming()) { $channel->wait(); } ~~~ 1. 運行代碼