Yii2 队列 shmilyzxt/yii2-queue 简单概述

shmilyzxt/yii2-queue 简单解释:

1.我用的yii2高级版,我们从配置开始看代码,这里我用的是mysql队列,首先配置文件,我把queue配置项写在根目录common/config/main-local.php下的 components数组下,更改一下数据库配置.复制composer安装后复制

vendor/shmilyzxt/yii2-queue/jobs/jobs.sqlvendor/shmilyzxt/yii2-queue/failed/failed.sql

2个sql文件到数据库中建立队列数据表和执行任务失败时的数据表.

2.推送任务开始语法:/Yii::$app->queue->pushOn(new SendMial(),['email'=>'49783121@qq.com','title'=>'test','content'=>'email test'],'email'); 我们到vendor/shmilyzxt/queue/queues/DatabaseQueue.php去看看代码,pushOn()方法写在了DatabaseQueue类的父类vendor/shmilyzxt/queue/base/Queue.php中:

//入队列public function pushOn($job, $data = '', $queue = null) { //canPush 检查队列是否已达最大任务量 if ($this->canPush()) {   //beforePush 入队列前的事件  $this->trigger(self::EVENT_BEFORE_PUSH);   //入队列  $ret = $this->push($job, $data, $queue);  //afterPush 入队列后的事件  $this->trigger(self::EVENT_AFTER_PUSH);  return $ret; } else {  throw new /Exception("max jobs number exceed! the max jobs number is {$this->maxJob}"); } }

注释:这里最好去看看yii2 event事件类,http://www.digpage.com/event.html

关于入队列: $this->push($job, $data, $queue);,这里在配合queue类文件查看,相关函数跳转,处理一下数据记录到数据库中.(函数走向:getQueue()-->createPayload()-->pushToDatabase()),pushOn()最终返回数据插入数据库的结果,成功$ret是1.

3.后台运行命令处理队列,例:php ./yii worker/listen default 10 128 3 0 其中default是队列的名称,上面推送了一个email队列 应该改为email.

启动命令后,我们来看代码:首先执行:WorkerController控制器 actionListen方法,我们跟着代码进入到 vendor/shmilyzxt/queue/Worker.php -- listen方法中,这里其实就是一直在循环,执行操作队列的任务:

/** * 启用一个队列后台监听任务 * @param Queue $queue * @param string $queueName 监听队列的名称(在pushon的时候把任务推送到哪个队列，则需要监听相应的队列才能获取任务) * @param int $attempt 队列任务失败尝试次数，0为不限制 * @param int $memory 允许使用的最大内存 * @param int $sleep 每次检测的时间间隔 */ public static function listen(Queue $queue, $queueName = 'default', $attempt = 10, $memory = 512, $sleep = 3, $delay = 0){ while (true){  try{  //DatabaseQueue从数据库队列取出一个可用任务(实例),并且更新任务  $job = $queue->pop($queueName);  }catch (/Exception $e){  throw $e;  continue;  }  if($job instanceof Job){  //判断执行错误的次数是否大于传入的执行次数  if($attempt > 0 && $job->getAttempts() > $attempt){   $job->failed();  }else{   try{   //throw new /Exception("test failed");   $job->execute();   }catch (/Exception $e){   //执行失败,判断是否被删除,重新入队   if (! $job->isDeleted()) {    $job->release($delay);   }   }  }  }else{  self::sleep($sleep);  }  if (self::memoryExceeded($memory)) {  self::stop();  } } }