tp6 RabbitMQ

1、composer 安装 AMQP 扩展

composer require php-amqplib/php-amqplib

2、RabbitMQ 配置

 在 config 目录下创建 rabbitmq.php 文件

<?php
return ['host'=>'','port'=>'5672','user'=>'','password'=>'','vhost'=>'','exchange_name' => '','queue_name' => '','route_key' => '','consumer_tag' => '',
];

3、生产者代码

app目录下创建Producer.php

<?phpnamespace app;use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;class Producer
{private $connection;private $channel;private $mq_config;public function __construct(){$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'],'itcast');//创建通道$this->channel = $this->connection->channel();}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']);$messageBody = json_encode($data);//将要发送数据变为json字符串/*** 创建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、消费者代码

app目录下创建Consumer.php

<?phpnamespace app;use app\index\controller\ApiCommunity;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use think\db\exception\PDOException;
use think\facade\Log;class Consumer
{private $connection;private $channel;private $mq_config;public function __construct(){$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['vhost']);//创建通道$this->channel = $this->connection->channel();}/*** @param $channel* @param $connection* 关闭进程*/function shutdown($channel, $connection){$channel->close();$connection->close();}/*** @param $message* 消息处理*/function process_message($message){//消息处理逻辑echo $message->body . "\n";if ($message->body !== 'quit') {$obj = json_decode($message->body);if (!isset($obj->id)) {Log::write("error data:" . $message->body, 2);} else {try {Log::write("data:" . json_encode($message));//消息处理} catch (\Think\Exception  $e) {Log::write($e->getMessage(), 2);Log::write(json_encode($message), 2);} catch (PDOException $pe) {Log::write($pe->getMessage(), 2);Log::write(json_encode($message), 2);}}}// 手动确认ack，确保消息已经处理$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']);}}/*** @throws \ErrorException* 启动** nohup php index.php index/Message_Consume/start &*/public function start(){// 设置消费者（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'], $this->mq_config['consumer_tag'], false, false, false, false, array($this, 'process_message'));register_shutdown_function(array($this, 'shutdown'), $this->channel, $this->connection);while (count($this->channel->callbacks)) {$this->channel->wait();}}
}

5、创建自定义命令

php think make:command Consumer

在项目跟目录执行以下命令，会自动生成 在 command 目录生成 Consumer 控制器 

<?php
declare (strict_types = 1);namespace app\command;use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;class Consumer extends Command
{protected function configure(){// 指令配置$this->setName('consumer')->setDescription('the consumer command');}protected function execute(Input $input, Output $output){// 指令输出$output->writeln('consumer');$consumer = new \app\Consumer();
//        $consumer->process_message(11)$consumer->start();}
}

config/console.php 代码增加如下：

// 指令定义
'commands' => ['consumer' => 'app\command\Consumer',
],

6、命令

消费者命令

php think consumer

 生产者执行命令

$producer = new Producer();
$data = ['message' => "发送的消息内容"
];
$producer->send($data);