>[info] ACK（confirm機制） **什么是 Confirm 機制?** 概念： * Pro發送消息到Broker,Broker接收到消息后,產生回送響應 Pro中有一個Confirm Listener異步監聽響應應答 步驟： * 消息的確認 Pro投遞消息后,如果Broker收到消息,則會給Pro一個應答 * Pro接收應答 用來確定這條消息是否正常地發送到Broker,該法也是消息可靠性投遞的核心保障! ***** **Confire 機制流程圖**  ***** **實現Confirm機制** 1\. 在channel上開啟確認模式：$channel->confirm\_select(); 2. 在channel上添加監聽：$channel->wait\_for\_pending\_acks(); 監聽成功和失敗的返回結果，根據具體的結果對消息進行重新發送、或記錄日志等后續處理。 ***** **ack 機制代碼：amqp_publisher_with_confirms.php** ~~~ <?php use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Exchange\AMQPExchangeType; use PhpAmqpLib\Message\AMQPMessage; include(__DIR__ . '/config.php'); $exchange = 'someExchange'; $connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST); $channel = $connection->channel(); //保存訂單信息到數據庫 //保存訂單信息到數據庫的代碼; //是否將信息成功推送給消息隊列(is_send_succ) //推送成功 $channel->set_ack_handler( function (AMQPMessage $message) { //update 訂單表 set is_send_succ=ture echo "Message acked with content " . $message->body . PHP_EOL; } ); //推送失敗 $channel->set_nack_handler( function (AMQPMessage $message) { //update 訂單表 set is_send_succ=false echo "Message nacked with content " . $message->body . PHP_EOL; } ); /* * bring the channel into publish confirm mode. * if you would call $ch->tx_select() before or after you brought the channel into this mode * the next call to $ch->wait() would result in an exception as the publish confirm mode and transactions * are mutually exclusive */ $channel->confirm_select(); /* name: $exchange type: fanout passive: false // don't check if an exchange with the same name exists durable: false // the exchange won't survive server restarts auto_delete: true //the exchange will be deleted once the channel is closed. */ $channel->exchange_declare($exchange, AMQPExchangeType::FANOUT, false, false, true); $i = 1; $msg = new AMQPMessage($i, array('content_type' => 'text/plain')); $channel->basic_publish($msg, $exchange); /* * watching the amqp debug output you can see that the server will ack the message with delivery tag 1 and the * multiple flag probably set to false */ $channel->wait_for_pending_acks(); while ($i <= 11) { $msg = new AMQPMessage($i++, array('content_type' => 'text/plain')); $channel->basic_publish($msg, $exchange); } /* * you do not have to wait for pending acks after each message sent. in fact it will be much more efficient * to wait for as many messages to be acked as possible. */ $channel->wait_for_pending_acks(); $channel->close(); $connection->close(); ~~~ **運行結果：**