手机付费咨询网站建设,长春做公司网站的,成都网络推广,六安服装网站建设地址php8.0 使用 rabbitmq 要使用 3.6版本以上的#xff0c; 并且还要开启 php.ini中的 socket 扩展 php think make:command SimpleMQProduce //创建一个生产者命令行 php think make:command SimpleMQConsumer //创建一个消费者命令行 代码中的消息持久化的说明
RabbitMQ 消息持…php8.0 使用 rabbitmq 要使用 3.6版本以上的 并且还要开启 php.ini中的 socket 扩展 php think make:command SimpleMQProduce //创建一个生产者命令行 php think make:command SimpleMQConsumer //创建一个消费者命令行 代码中的消息持久化的说明
RabbitMQ 消息持久化 持久化是为提高rabbitmq消息的可靠性防止在异常情况(重启关闭宕机)下数据的丢失。设置完队列和消息的持久化并不能完全保证消息不会丢失。尽管它告诉 RabbitMQ 将消息保存到磁盘但当 RabbitMQ 接受消息但尚未保存消息时仍有一个较短的时间窗口。另外 RabbitMQ 不会对每条消息都执行 fsync2 – 它可能只是保存到缓存中而不是真正写入磁盘。持久性保证并不强但对于我们的简单任务队列来说已经足够了。如果您需要更强的保证则可以使用发布者确认RabbitMQ 消息确认机制
生产者代码
?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 [namehuang.$i,age$i,sexman.$i];$msg new AMQPMessage(json_encode($message),[delivery_modeAMQPMessage::DELIVERY_MODE_PERSISTENT //使消息持久化]);$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);//从队列中消费数据/*** 参数说明* params queue 队列名称* params consumer_tag 消费者标签* params no_local AMQP标准 RabbitMQ没有实现 默认为false* params no_ack 收到消息后是否不需要回复 如果为true表示不回复 一般是要回复的 设置为 false* params exclusive 排他消费即这个队列只能由一个消费者来消费* params nowait 相当于一个异步在执行完成之后返回结果不堵塞* params callback 处理消息的回调方法*/$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(队列连接失败);}}
}
