簡述Yii2隊列shmilyzxt/yii2-queue

本文主要介紹了Yii2 隊列 shmilyzxt/yii2-queue 的簡單概述，需要的朋友可以參考下。希望對大家有幫助。

shmilyzxt/yii2-queue 簡單解釋:

1.我用的yii2高級版,我們從設定開始看程式碼,這裡我用的是mysql佇列,先設定檔,我把queue配置項目寫在根目錄common\config\main-local.php下的 components陣列下,更改一下資料庫配置.複製composer安裝後複製

vendor\shmilyzxt\yii2-queue\jobs\jobs.sql
vendor\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();
  }
 }
 }

註釋:在$queue->pop($queueName);是vendor\shmilyzxt\queue\queues\DatabaseQueue.php方法內使用交易執行SQL,並且建立vendor \shmilyzxt\queue\jobs\DatabaseJob.php的實例

 //取出一个任务
 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;
 }

至於:$job->execute();是DatabaseJob繼承父類別Job執行的,順著程式碼找下去是yii\base\Component trigger執行的事件,

/**
 * 执行任务
 */
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();
  }
 }

最後到了執行的SendMail類別的handle($job,$data ),在這裡就是推送到佇列的物件和資料,接著就是我們的處理邏輯了.

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

相關推薦：

Yii2整合迅搜實現高效能中文分詞檢索

#Yii解決DeleteAll連表刪除錯誤問題

######################################### ########Yii如何過濾不良程式碼##########

以上是簡述Yii2隊列shmilyzxt/yii2-queue的詳細內容。更多資訊請關注PHP中文網其他相關文章！