tp8 使用rabbitMQ

2023-11-23 13:01

文章标签 使用 rabbitmq tp8

本文主要是介绍tp8 使用rabbitMQ，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

php8.0 使用 rabbitmq 要使用 3.6版本以上的，并且还要开启 php.ini中的 socket 扩展

在这里插入图片描述
php think make:command SimpleMQProduce //创建一个生产者命令行
php think make:command SimpleMQConsumer //创建一个消费者命令行

在这里插入图片描述

生产者代码

<?php
declare (strict_types = 1);namespace app\command;use ba\Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;//本类是模式中的 生产者 produce
class SimpleMQProduce extends Command
{protected function configure(){// 指令配置$this->setName('SimpleMQ')->setDescription('这是一个队列的 Hello模式 (最简单的应用模式)，使用了默认的交换机，只需要建一个队列就可以了');}protected function execute(Input $input, Output $output){//获取连接$connection = $this->connectRabbitMQ();//获取连接的通道$channel = $connection->channel();//直接创建一个队列/*** 关于 queue_declare参数的说明* params  queue  队列的名称* params  passive 是否消极的声明队列，如果存在，就把队列的信息返回， 如果没有就抛出错误，（是的， 你没看错，这个参数很鸡肋，所以一般为 false）* params  durable 是否持久化，意思是说就算队列服务持了， 也不会丢失队列* params  exclusive  是否排外，如果设置为true ,表示只有本次连接中的channel 可以访问，其它channel 是不可以访问的* params  auto_delete  设置是否自动删除。为true 则设置队列为自动删除。自动删除的前提是, 至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除* params  nowait 相当于做一个异步版的声明， 如果设置成true, 就是说方法调用完就结束，也不用等待创建队列是否成功，一般也设为false*/$channel->queue_declare("hello",false,false,false,false,false);for ($i = 0; $i < 20; $i++) {$message = ["name"=>"huang".$i,"age"=>$i,"sex"=>"man".$i];$msg = new AMQPMessage(json_encode($message));$channel->basic_publish($msg,"",'hello');  //简单模式下，routing_key 和 队列名称是一样的sleep(1);}//生产者调用完成后要关闭资源$channel->close();}//这个方法是建立一个 rabbitMQ 的资源连接protected function connectRabbitMQ(){try{$connection = new AMQPStreamConnection("192.168.3.228",'5672',"admin","123456");return $connection;}catch(Exception $e){throw new Exception("队列连接失败");}}}

消费者代码

<?php
declare (strict_types = 1);namespace app\command;use ba\Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;class SimpleMQConsumer extends Command
{protected function configure(){// 指令配置$this->setName('simplemqconsumer')->setDescription('这是rabbitMQ的简单模式的消费者代码');}protected function execute(Input $input, Output $output){//创建一个资源连接$connection = $this->connectRabbitMQ();$channel = $connection->channel();//定义一个处理函数$callback = function($msg){$body = $msg->body;$bodyArr = json_decode($body,true);echo $bodyArr["name"] ."--".$bodyArr["age"]. "--".$bodyArr["sex"];};//声明队列，这个函数的 passive ，如果有不返回这个队列的信息，如果没有就抛出异常$channel->queue_declare("hello",false,false,false,false,false);//从队列中消费数据$channel->basic_consume("hello",'',false,true,false,false,$callback);//等待队列执行完成while(count($channel->callbacks)){$channel->wait();}}//这个方法是建立一个 rabbitMQ 的资源连接protected function connectRabbitMQ(){try{$connection = new AMQPStreamConnection("192.168.3.228",'5672',"admin","123456");return $connection;}catch(Exception $e){throw new Exception("队列连接失败");}}
}