>[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(); ~~~ **运行结果：**