vendor\shmilyzxt\yii2-queue\jobs\jobs.sql
vendor\shmilyzxt\yii2-queue\failed\failed.sql

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

/**
 * 啟用一個(gè)隊(duì)列后臺(tái)監(jiān)聽(tīng)任務(wù)
 * @param Queue $queue
 * @param string $queueName 監(jiān)聽(tīng)隊(duì)列的名稱(在pushon的時(shí)候把任務(wù)推送到哪個(gè)隊(duì)列,則需要監(jiān)聽(tīng)相應(yīng)的隊(duì)列才能獲取任務(wù))
 * @param int $attempt 隊(duì)列任務(wù)失敗嘗試次數(shù),0為不限制
 * @param int $memory 允許使用的最大內(nèi)存
 * @param int $sleep 每次檢測(cè)的時(shí)間間隔
 */
 public static function listen(Queue $queue, $queueName = 'default', $attempt = 10, $memory = 512, $sleep = 3, $delay = 0){
 while (true){
  try{
  //DatabaseQueue從數(shù)據(jù)庫(kù)隊(duì)列取出一個(gè)可用任務(wù)(實(shí)例),并且更新任務(wù)
  $job = $queue->pop($queueName);
  }catch (\Exception $e){
  throw $e;
  continue;
  }
  if($job instanceof Job){
  //判斷執(zhí)行錯(cuò)誤的次數(shù)是否大于傳入的執(zhí)行次數(shù)
  if($attempt > 0 && $job->getAttempts() > $attempt){
   $job->failed();
  }else{
   try{
   //throw new \Exception("test failed");
   $job->execute();
   }catch (\Exception $e){
   //執(zhí)行失敗,判斷是否被刪除,重新入隊(duì)
   if (! $job->isDeleted()) {
    $job->release($delay);
   }
   }
  }
  }else{
  self::sleep($sleep);
  }
  if (self::memoryExceeded($memory)) {
  self::stop();
  }
 }
 }

 //取出一個(gè)任務(wù)
 public function pop($queue = null)
 {
 $queue = $this->getQueue($queue);
 if (!is_null($this->expire)) {
  //$this->releaseJobsThatHaveBeenReservedTooLong($queue);
 }
 $tran = $this->connector->beginTransaction();
 //判斷是否有一個(gè)可用的任務(wù)需要執(zhí)行
 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;
 }

/**
 * 執(zhí)行任務(wù)
 */
public function execute()
{
 $this->trigger(self::EVENT_BEFORE_EXECUTE, new JobEvent(["job" => $this, 'payload' => $this->getPayload()]));//beforeExecute 執(zhí)行任務(wù)之前的一個(gè)事件 在JobEvent中并沒(méi)有什么可執(zhí)行的代碼
 $this->resolveAndFire();//真正執(zhí)行的任務(wù)的方法
}
 /**
 * 真正任務(wù)執(zhí)行方法(調(diào)用hander的handle方法)
 * @param array $payload
 * @return void
 */
 protected function resolveAndFire()
 {
  $payload = $this->getPayload();
  $payload = unserialize($payload); //反序列化數(shù)據(jù)
  $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 {//執(zhí)行的`SendMail`類的`handle($job,$data)`方法
   $this->handler = $this->getHander($class);
   $this->handler->handle($this, $payload['data']);
  }
  //執(zhí)行完任務(wù)后刪除
  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即任務(wù)的數(shù)據(jù),你拿到任務(wù)數(shù)據(jù)后就可以執(zhí)行發(fā)郵件了
  //TODO 發(fā)郵件
 }