composer.json

{
  "require": {
    "php-amqplib/php-amqplib": "2.6.1"
  }
}

product.php

/**
 * Created by PhpStorm.
 * User: xiaowen
 * Date: 2018/12/10
 * Time: 5:24 PM
 */
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitProducter{
    private $exchange_name = "exchange_demo";
    private $routing_key   = "routingKey_demo";
    private $queue         = "queue_demo";
    private $ip_address    = "localhost";
    private $port          = 5672;
    public function send()
    {
        //创建连接
        $connection = new AMQPStreamConnection($this->ip_address, $this->port, 'guest', 'guest');
        //创建信道
        $channel = $connection->channel();
        //创建一个type=direct 持久化 非自动删除的交换器 passive只是是确认队列是否存在 而不会去创建
        $channel->exchange_declare($this->exchange_name, "direct",false, true, false);
        //创建一个持久化非拍他非自动删除的队列
        $channel->queue_declare($this->queue, false, true, false, false);
        //将交换器和队列通过路由键绑定
        $channel->queue_bind($this->queue, $this->exchange_name, $this->routing_key);
        //发送一条持久化消息
        $msg = new AMQPMessage('Hello World!');
        $channel->basic_publish($msg, $this->exchange_name, $this->routing_key);
        $channel->close();
        $connection->close();
    }
}
$product = new RabbitProducter();
$product->send();
echo 'ok';

consumer.php

/**
 * Created by PhpStorm.
 * User: xiaowen
 * Date: 2018/12/10
 * Time: 5:24 PM
 */
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

class RabbitConsumer{
    private $exchange_name = "exchange_demo";
    private $routing_key   = "routingKey_demo";
    private $queue         = "queue_demo";
    private $ip_address    = "localhost";
    private $port          = 5672;
    public function send()
    {
        //创建连接
        $connection = new AMQPStreamConnection($this->ip_address, $this->port, 'guest', 'guest');
        //创建信道
        $channel = $connection->channel();
        //设置客户端最多接收未被ack的消息的个数
        $channel->basic_qos(null, 1, null);
        $channel->queue_declare($this->queue, false, true, false, false);
        $callback = function($msg) {
            echo " [x] Received ", $msg->body, "\n";
        };
        $channel->basic_consume($this->queue, '', false, true, false, false, $callback);
        while(count($channel->callbacks)) {
            $channel->wait();
        }
    }
}
$consumer = new RabbitConsumer();
$consumer->send();
echo 'ok';