thinkphp3.2使用PhpAmqpLib（RabbitMq

原文地址：https://www.jianshu.com/p/439c155cc04b
原文地址：https://www.helloweba.net/php/625.html

1.安装php-amqplib

第一种：composer require php-amqplib/php-amqplib
第二种：下载地址 https://github.com/php-amqplib/php-amqplib

2.在控制器中引入就可以使用了

<?php
/**
 * RabbitMq 生产者消费者demo
 */
namespace Wxuser\Controller;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Think\Controller;

class RabbitMqController extends Controller{
    const consumerTag = '';//消费者标签 可为空
    const exchange = 'fanout'; // 交换机类型
    const queue = 'transfer_recordQueue';//队列名
    
    /**
     * 推入消息到队列中
     */
    public function pushMessage()
    {
        //测试数据
        $data = ["test" =>32132132];
        // 连接rabbitMQ
        $connection = new AMQPStreamConnection('xxxx.xxx.xxx', 5672, 'admin', '88888', '/');
                //$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);        // 开启一个信道
        $channel = $connection->channel();
        // 声明一个队列
        // queue 队列名
        // passive 检测队列是否存在  true 只检测不创建 false 创建
        // durable 是否持久化队列 true 为持久化
        // exclusive 私有队列 不允许其它用户访问  设置true 将会变成私有
        // auto_delete  当所有消费客户端连接断开后，是否自动删除队列
        $channel->queue_declare(self::queue, false, true, false, false);
    
        // exchange 交换机名称
        // type 交换器类型
        // passive 检测交换机是否存在   true 只检测不创建 false 创建
        // durable 是否持久化队列 true 为持久化
        // auto_delete  当所有绑定队列都不在使用时，是否自动删除交换器 true：删除false：不删除
        $channel->exchange_declare(self::exchange, 'direct', false, true, false);
        // 绑定队列和交换机
        $channel->queue_bind(self::queue, self::exchange);
        // 写入队列的消息
        $messageBody = json_encode($data) ;
        // 消息内容
        // delivery_mode  投递模式  delivery mode 设置为 2标记持久化
        $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
        // $message 消息内容
        // $exchange  交换器名称
        // routing_key   路由键 (routing key)  主题交换机会用到
        $channel->basic_publish($message, self::exchange,'');
        // 关闭信道
        $channel->close();
        //关闭 amqp 连接
        $connection->close();
        return "ok";
    }
    
    
    function shutdown($channel, $connection)
    {
        $channel->close();
        $connection->close();
    }
    
    function process_message($message)
    {
        if ($message->body !== 'quit') {
            $obj = json_decode($message->body);
            $message->body;//拿到的数据
            dump($message->body);
            #业务操作
//          if (!isset($obj->id)) {
//              //  消费成功会在 日志里面写入一条数据
//              Log::info("error data111111111111111:" . $message->body, 2);
//          } else {
//              try {
//                  Log::info("data:" . json_encode($message));
//              } catch (\Think\Exception  $e) {
//                  Log::info($e->getMessage(), 2);
//                  Log::info(json_encode($message), 2);
//              } catch (\PDOException $pe) {
//                  Log::info($pe->getMessage(), 2);
//                  Log::info (json_encode($message), 2);
//              }
//          }
        }
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        // Send a message with the string "quit" to cancel the consumer.
        if ($message->body === 'quit') {
            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
        }
    }
    
    /**
     * 启动
     *
     * @return \think\Response
     */
    public function start()
    {
        $connection = new AMQPStreamConnection('xxxx.xxx.xxx', 5672, 'admin', '88888', '/');
                //$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);
        $channel = $connection->channel();
        $channel->queue_declare(self::queue, false, true, false, false);
        $channel->exchange_declare(self::exchange, 'direct', false, true, false);
        $channel->queue_bind(self::queue, self::exchange);
        // queue   队列名称
        // consumer_tag  消费者标签
        // no_ack  在设置了 no_ack=false 的情况下）只要 consumer 手动应答了 Basic.Ack ，就算其“成功”处理了
        // no_ack=true （此时为自动应答）
        // exclusive  是否是私有队列 设置true 将会变成私有
        // callback = null, 回调函数
        $channel->basic_consume(self::queue, self::consumerTag, false, false, false, false, array($this, 'process_message'));
        // 不管你的php代码执行是否成功，最后都会执行 shutdown方法，关闭信道和连接
        register_shutdown_function(array($this, 'shutdown'), $channel, $connection);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
    }
}

3.执行命令，就可以监听消费队列了

linux环境 守护进程启动
nohup php index.php Wxuser/RabbitMq/start &
windows环境
php index.php Wxuser/RabbitMq/start