扩展包:

composer require php-amqplib/php-amqplib

生产者(发送):

<?phprequire_once __DIR__ . '/vendor/autoload.php';use PhpAmqpLib\Connection\AMQPStreamConnection;use PhpAmqpLib\Message\AMQPMessage;$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');$channel = $connection->channel();$channel->queue_declare('hello', false, false, false, false);$msg = new AMQPMessage('你好!');$channel->basic_publish($msg, '', 'hello');$channel->close();$connection->close();

消费者(接收):

<?phprequire_once  'vendor/autoload.php';use PhpAmqpLib\Connection\AMQPStreamConnection;$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');$channel = $connection->channel();$channel->queue_declare('hello', false, false, false, false);$callback = function($msg) {
  echo " [x] Received ", $msg->body, "\n";
    // 手动确认ack，确保消息已经处理
  $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
  if ($msg->body === 'quit') {
     $msg->delivery_info['channel']->basic_cancel($msg->delivery_info['consumer_tag']);
  }};$channel->basic_consume('hello', '', false, false, false, false, $callback);while(count($channel->callbacks)) {
    $channel->wait();}