easy-swoole-rabbitmq

EasySwoole 框架的 RabbitMQ 队列插件，基于 php-amqplib/php-amqplib

支持的交换器类型

已支持 direct topic fanout

安装

composer require lys/easy-swoole-rabbitmq

示例:

我们创建一个MqQueueProcess.php, 消费者进程

<?php namespace AppUtility; use EasySwooleComponentProcessAbstractProcess; use EasySwooleEasySwooleLogger; use EasySwooleRabbitMqMqQueue; use EasySwooleRabbitMqMqJob; class MqQueueProcess extends AbstractProcess { protected function run($arg) { go(function () { $MqQueue = MqQueue::getInstance()->refreshConnect(); $MqQueue->consumer()->setConfig($exchange = 'kd_sms_send_ex', $routingKey = 'hello', $mqType = 'direct', $queueName = 'hello')->listen(function(MqJob $obj) { echo " [x] Received ", $obj->getJobData(), "n"; Logger::getInstance()->log('log level info' . var_export($obj->getJobData(), true), Logger::LOG_LEVEL_INFO, 'DEBUG');//记录info级别日志//例子后面2个参数默认值 var_dump($obj->getJobData(),'MqQueueProcess'); //return; //使用return 终止执行下面的代码 //return true; //使用return true终止执行下面的代码 echo 11111; //return false; //return false消息回滚,所以请注意，不要随意使用return false }); }); } }

修改EasySwooleEvent.php，在mainServerCreate中添加如下代码

<?php namespace EasySwooleEasySwoole; use EasySwooleComponentTimer; use EasySwooleEasySwooleSwooleEventRegister; use EasySwooleEasySwooleAbstractInterfaceEvent; use EasySwooleHttpRequest; use EasySwooleHttpResponse; use EasySwooleRabbitMqMqQueue; use EasySwooleRabbitMqMqJob; use EasySwooleRabbitMqRabbitMqQueueDriver; use AppUtilityMqQueueProcess; class EasySwooleEvent implements Event { public static function initialize() { // TODO: Implement initialize() method. date_default_timezone_set('Asia/Shanghai'); } public static function mainServerCreate(EventRegister $register) { $driver = new RabbitMqQueueDriver('127.0.0.1', 5672, 'test', 'test',"/"); MqQueue::getInstance($driver); $processConfig= new EasySwooleComponentProcessConfig(); $processConfig->setProcessGroup('Test');//设置进程组 $processConfig->setArg(['a'=>123]);//传参 $processConfig->setRedirectStdinStdout(false);//是否重定向标准io $processConfig->setPipeType($processConfig::PIPE_TYPE_SOCK_DGRAM);//设置管道类型 $processConfig->setEnableCoroutine(true);//是否自动开启协程 $processConfig->setMaxExitWaitTime(3);//最大退出等待时间 $processConfig->setProcessName('MqQueueProcessComposer'); EasySwooleEasySwooleServerManager::getInstance()->addProcess(new MqQueueProcess($processConfig)); } public static function onRequest(Request $request, Response $response): bool { // TODO: Implement onRequest() method. return true; } public static function afterRequest(Request $request, Response $response): void { // TODO: Implement afterAction() method. } }

生产者投递消息

<?php namespace AppUtility; use EasySwooleRabbitMqMqQueue; use EasySwooleRabbitMqMqJob; class MqComposer{ public static function push(){ $job = new MqJob(); $job->setJobData('composer hello word'.date('Y-m-d H:i:s', time())); $res = MqQueue::getInstance()->producer()->setConfig($exchange = 'kd_sms_send_ex',$routingKey = 'hello',$mqType = 'direct', $queueName = 'hello')->push($job); if($res){ var_dump('发布成功'); }else{ var_dump('发布失败'); } //主动关闭链接 /* MqQueue::getInstance()->closeConnection(function (Exception $e){ //这边是主动关闭异常处理 }); */ } }

更多请关注本人的博客

https://www.developzhe.com/single200.html

Page visitor counter