1. composer安装amqp包

composer require php-amqplib/php-amqplib

2. 新建配置文件

config/rabbit_mq.php

return [
    'host'=>'127.0.0.1',
    'port'=>'5672',
    'user'=>'test',
    'password'=>'test',
 
];

3. 封装生产者

消费者和生产者的大多数代码都是一样的，在tp框架目录下的extend/下创建RabbitMqUtilsOri类，代码如下:

<?php

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMqUtilsOri
{

    private $connection;
    private $channel;
    private $mq_config;

    /**
     * @param $exchange_name
     * @param $route_key
     * @param $queue_name
     */
    public function __construct($exchange_name,$route_key,$queue_name)
    {
        // 获取配置文件参数
        $this->mq_config = config('rabbit_mq');

        $this->connection = new AMQPStreamConnection(
            $this->mq_config['host'],
            $this->mq_config['port'],
            $this->mq_config['user'],
            $this->mq_config['password']
        );
        // 声明 rabbitmq的 交换机 路由key 队列
        $this->mq_config['exchange_name'] = $exchange_name;

        $this->mq_config['route_key'] = $route_key;

        $this->mq_config['queue_name'] = $queue_name;

        //创建通道
        $this->channel = $this->connection->channel();
    }

    /**
     * 发送消息
     * @param $data
     */
    public function send($data)
    {
        /**
         * 创建队列(Queue)
         * name: hello         // 队列名称
         * passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建
         * durable: true       // 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失;设置true,则代表是一个持久化的队列，服务重启后也会存在，因为服务会把持久化的queue存放到磁盘上当服务重启的时候，会重新加载之前被持久化的queue
         * exclusive: false    // 是否排他，指定该选项为true则队列只对当前连接有效，连接断开后自动删除
         * auto_delete: false // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
         *
         */
        $this->channel->queue_declare($this->mq_config['queue_name'], false, true, false, false);

        /**
         * 创建交换机(Exchange)
         * name: vckai_exchange// 交换机名称
         * type: direct        // 交换机类型，分别为direct/fanout/topic，参考另外文章的Exchange Type说明。
         * passive: false      // 如果设置true存在则返回OK，否则就报错。设置false存在返回OK，不存在则自动创建
         * durable: false      // 是否持久化，设置false是存放到内存中的，RabbitMQ重启后会丢失
         * auto_delete: false  // 是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除
         */
        $this->channel->exchange_declare($this->mq_config['exchange_name'], AMQPExchangeType::DIRECT, false, true, false);

        // 绑定消息交换机和队列
        $this->channel->queue_bind($this->mq_config['queue_name'], $this->mq_config['exchange_name'],$this->mq_config['route_key']);
        // 将要发送数据变为json字符串  去除反斜杠 中文不转译
        $messageBody = str_replace("\\/", "/", json_encode($data,
            JSON_UNESCAPED_UNICODE));

        /**
         * 创建AMQP消息类型
         * delivery_mode 消息是否持久化
         * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT  不持久化
         * AMQPMessage::DELIVERY_MODE_PERSISTENT      持久化
         */
        $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));

        /**
         * 发送消息
         * msg: $message            // AMQP消息内容
         * exchange: vckai_exchange // 交换机名称
         * routing_key: hello       // 路由key
         */
        $this->channel->basic_publish($message, $this->mq_config['exchange_name'], $this->mq_config['route_key']);

        //关闭连接
        $this->stop();

    }

    //关闭进程
    public function stop()
    {
        $this->channel->close();
        $this->connection->close();
    }


}

4. 生产者发送消息

这里模拟一个创建用户的队列消息，控制器中创建一个发消息的方法：

public function sendMsg(Request $request){
        $msg = ["name" => "张三", "age" => 22];
        // RabbitMqUtilsOri是上面自己封装的类，第一个参数是交换机名称，第二个参数是队列名称，第三个参数是路由键名称，这里采用的是TOPIC模型，可以匹配字符串
        $mq = new \RabbitMqUtilsOri("user_ex","user_que","user.insert");
        $mq->sendMsg($msg);
        return json(["code"=>200,"msg"=>"success"]);
}

5. rabbitmq的控制界面查看效果

(1) rabbitmq控制面板

打开rabbitmq的控制界面，不知道怎么操作的回到本文顶部环境安装有链接

(2) 发送消息

现在队列是干净的，现在开始访问这个发消息的方法。请求接口返回

(3) 查看rabbitmq的交换机

然后看下控制界面，首先交换机下有一个刚才我们创建的user，并且是持久化的

(4) 查看rabbitmq队列

(5) 查看数据

队列中也有了一条消息，这条消息是ready状态，还没被消费，可以看下，就是我们刚才的那条

6. 消费者接收消息

(1) 消费者运行

消费者一般都是cli模式在xshell下运行一个php 文件，如果要用TP框架的话，需要进入项目的public目录，运行

php index.php 模块/控制器/方法 &

&代表守护进程运行
然后是消费者的代码，实际业务中，消费者在接到消息的时候，肯定要根据不同的消息做不同的逻辑处理，比如我们这就是需要把消息中的user信息插入到数据库，所以如果直接把消息处理的逻辑写到消费者代码中耦合度太高，相关资料中呢都是这样写的

(2) 封装消费者

<?php
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use think\db\exception\PDOException;
use think\facade\Log;

class RabbitmqConsumer
{
    private $connection;
    private $channel;
    private $mq_config;

    public function __construct($exchange_name,$route_key,$queue_name)
    {
        $this->mq_config = config('rabbit_mq');
        $this->connection = new AMQPStreamConnection(
            $this->mq_config['host'],
            $this->mq_config['port'],
            $this->mq_config['user'],
            $this->mq_config['password']
        );

        $this->mq_config['exchange_name'] = $exchange_name;

        $this->mq_config['route_key'] = $route_key;

        $this->mq_config['queue_name'] = $queue_name;
        //创建通道
        $this->channel = $this->connection->channel();
    }

    /**
     * @param $channel
     * @param $connection
     * 关闭进程
     */
    function shutdown($channel, $connection)
    {
        $channel->close();
        $connection->close();
    }


    /**
     * @throws \ErrorException
     * 启动
     *
     * nohup php index.php index/Message_Consume/start &
     */
    public function start($parent)
    {
        // 设置消费者（Consumer）客户端同时只处理一条队列
        // 这样是告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个消费者（Consumer），直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ就会把消息分发给下一个空闲的消费者（Consumer）。
        //消费者端要把自动确认autoAck设置为false，basic_qos才有效。
        //$this->channel->basic_qos(0, 1, false);

        // 同样是创建路由和队列，以及绑定路由队列，注意要跟producer（生产者）的一致
        // 这里其实可以不用设置，但是为了防止队列没有被创建所以做的容错处理
        $this->channel->queue_declare($this->mq_config['queue_name'], false, true, false, false);
        $this->channel->exchange_declare($this->mq_config['exchange_name'], AMQPExchangeType::DIRECT, false, true, false);
        $this->channel->queue_bind($this->mq_config['queue_name'], $this->mq_config['exchange_name'], $this->mq_config['route_key']);

        /**
         *
         * queue: queue_name    // 被消费的队列名称
         * consumer_tag: consumer_tag // 消费者客户端身份标识，用于区分多个客户端
         * no_local: false      // 这个功能属于AMQP的标准，但是RabbitMQ并没有做实现
         * no_ack: true         // 收到消息后，是否不需要回复确认即被认为被消费
         * exclusive: false     // 是否排他，即这个队列只能由一个消费者消费。适用于任务不允许进行并发处理的情况下
         * nowait: false        // 不返回执行结果，但是如果排他开启的话，则必须需要等待结果的，如果两个一起开就会报错
         * callback: $callback  // 回调逻辑处理函数
         *
         */
        $this->channel->basic_consume(
            $this->mq_config['queue_name'],
            false,
            false, false, false, false,
            $parent
        );

        register_shutdown_function(array($this, 'shutdown'), $this->channel, $this->connection);

        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }
}

(3) 创建请求方法

创建一个Message的类也好，控制器也行，当前是创建 方法 代码如下

    public function receiveMsg()
    {
        $mq = new \RabbitmqConsumer("user","user","user");
        try {
            $mq->start([$this, "processMessage"]);
        } catch (\ErrorException $e) {
            return ['code'=>400,'消费失败'];
        }
    }

(4) 创建处理方法

一会接到消息让这个方法去处理，这个类下也可以写很多的方法，根据不同的消息处理不同的业务逻辑，下面是消费者代码:

    /**
     * @param $message
     * 消息处理
     */
    public function processMessage($message)
    {
        //消息处理逻辑
        echo $message->body . "\n";
        // 新建rabbitmq 日志信道，写入日志
        Log::channel('rabbitmq')->write(json_decode($message->body));

        // 手动确认ack，确保消息已经处理
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);

    }

(5) 运行脚本命令

当监听到消息时，会调用当前控制器中的processMessage方法,该方法中会向日志里面插入一条数据。运行代码，在/public下，运行

php index.php index/receiveMsg &

// 后续可以用Supervisor 部署 可以很方便地做到进程自动恢复的目的，而无需自己编写 shell 脚本来管理进程。

(6) 查看队列数据

先看下rabbitmq,刚才的消息被消费掉了

(7) 查看日志