/**
     * 普通的推送消息
     * @throws \Exception
     * User: cwh  DateTime:2021/8/20 0:12
     */
   public function send(){
       //连接rabbitmq
       $connection = new AMQPStreamConnection(‘localhost‘, 5672, ‘admin‘, ‘admin‘);
       $channel = $connection->channel();

       $queue_name = ‘normal‘;
       //声明一个队列
       // passive true 不会自动创建队列  false当队列不存在会自动创建队列
       // durable 消息持久化
       //Exclusive true 排他队列，false 非排他队列 排他队列，如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，
       //并在连接断开时自动删除。这里需要注意三点：其一，排他队列是基于连接可见的，同一连接的不同信道是可以同时访问同
       //一个连接创建的排他队列的。其二，“首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，
       //这个与普通队列不同。其三，即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的。
       //这种队列适用于只限于一个客户端发送读取消息的应用场景
       //  auto_delete 自动删除如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列
       $channel->queue_declare($queue_name, false, false, false, false);

       $msg = new AMQPMessage(‘Hello World!‘);
       //发送消息到队列中
       $channel->basic_publish($msg, ‘‘, $queue_name);

       echo " [x] Sent ‘Hello World!‘\n";

       $channel->close();
       $connection->close();
   }

    /**
     * 普通消费
     * @throws \ErrorException
     * User: cwh  DateTime:2021/8/20 0:58
     */
   public function receive(){
       $connection = new AMQPStreamConnection(‘localhost‘, 5672, ‘admin‘, ‘admin‘);
       $channel = $connection->channel();
       $queue_name = ‘normal‘;
       $channel->queue_declare($queue_name, false, false, false, false);

//       echo " [*] Waiting for messages. To exit press CTRL+C\n";
        //监听队列
       $channel->basic_consume($queue_name, ‘‘, false, true, false, false, [$this,‘callback‘]);

       while ($channel->is_open()) {
           $channel->wait();
       }

       $channel->close();
       $connection->close();
   }

    /**
     * 回调
     * @param $msg
     * User: cwh  DateTime:2021/8/20 1:00
     */
   public function callback($msg){
       logs("回调普通模式:".$msg->body,‘pushMessage‘);
   }

执行生产者 normal队列多了一条数据

执行消费者 发现数据被消费了

tp5.1 rabbitmq 实现简单队列