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("队列连接失败");
}
}
}