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

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

//入队列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}"); } }

/** * 启用一个队列后台监听任务 * @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();  } } }

 //取出一个任务 public function pop($queue = null) { $queue = $this->getQueue($queue); if (!is_null($this->expire)) {  //$this->releaseJobsThatHaveBeenReservedTooLong($queue); } $tran = $this->connector->beginTransaction(); //判断是否有一个可用的任务需要执行 if ($job = $this->getNextAvailableJob($queue)) {  $this->markJobAsReserved($job->id);  $tran->commit();  $config = array_merge($this->jobEvent, [  'class' => 'shmilyzxt/queue/jobs/DatabaseJob',  'queue' => $queue,  'job' => $job,  'queueInstance' => $this,  ]);  return /Yii::createObject($config); } $tran->commit(); return false; }

/** * 执行任务 */public function execute(){ $this->trigger(self::EVENT_BEFORE_EXECUTE, new JobEvent(["job" => $this, 'payload' => $this->getPayload()]));//beforeExecute 执行任务之前的一个事件 在JobEvent中并没有什么可执行的代码 $this->resolveAndFire();//真正执行的任务的方法} /** * 真正任务执行方法（调用hander的handle方法） * @param array $payload * @return void */ protected function resolveAndFire() {  $payload = $this->getPayload();  $payload = unserialize($payload); //反序列化数据  $type = $payload['type'];  $class = $payload['job'];  if ($type == 'closure' && ($closure = (new Serializer())->unserialize($class[1])) instanceof /Closure) {   $this->handler = $this->getHander($class[0]);   $this->handler->closure = $closure;   $this->handler->handle($this, $payload['data']);  } else if ($type == 'classMethod') {   $payload['job'][0]->$payload['job'][1]($this, $payload['data']);  } else if ($type == 'staticMethod') {   $payload['job'][0]::$payload['job'][1]($this, $payload['data']);  } else {//执行的`SendMail`类的`handle($job,$data)`方法   $this->handler = $this->getHander($class);   $this->handler->handle($this, $payload['data']);  }  //执行完任务后删除  if (!$this->isDeletedOrReleased()) {   $this->delete();  } }

public function handle($job,$data) {  if($job->getAttempts() > 3){   $this->failed($job);  }  $payload = $job->getPayload();  echo '<pre>';print_r($payload);  //$payload即任务的数据，你拿到任务数据后就可以执行发邮件了  //TODO 发邮件 }