Introduction à la file d'attente PHPyii2 shmilyzxt/yii2-queue

• Je trouve que c'est plutôt bien lors de l'utilisation. Je vous propose de jeter un oeil aux principes des files d'attente

• shmilyzxt/yii2-queueExplication simple :

  .

1. J'utilise la version avancée de yii2. Commençons par la configuration et regardons le code. Ici, j'utilise la file d'attente mysql. Tout d'abord, je configure le fichier <.> élément de configuration dans le répertoire racine. Sous le tableau queue sous commonconfigmain-local.php, modifiez la configuration de la base de données Copiez components et après l'installation, copiez le composer
vendorshmilyzxtyii2-queuejobsjobs.sql
vendorshmilyzxtyii2-queuefailedfailed.sql 2. sql dans la base de données pour créer la table de données de file d'attente et la table de données en cas d'échec de l'exécution de la tâche
   

2. Syntaxe de démarrage de la tâche push :

   Allons à Yii::$app->queue->pushOn(new SendMial(),['email'=>'49783121@qq.com','title'=>'test','content'=>'email test'],'email'); pour y jeter un œil. le code. La méthode vendorshmilyzxtqueuequeuesDatabaseQueue.php est écrite dans la classe pushOn() Classe parent DatabaseQueue : vendorshmilyzxtqueuebaseQueue.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}");
        }
    }

Remarque : C'est le cas. Il est préférable de jeter un œil à la classe d'événements yii2 ici

À propos de la mise en file d'attente :

, ici nous coopérons avec le fichier de classe $this->push($job, $data, $queue); pour afficher, accéder aux fonctions associées, traiter les données et les enregistrer. dans la base de données. (Direction de la fonction : queue), getQueue()-->createPayload()-->pushToDatabase() renvoie finalement le résultat de l'insertion des données dans la base de données, Succès pushOn() est 1. $ret

3. Exécutez la file d'attente de traitement des commandes en arrière-plan. , par exemple :

où php ./yii worker/listen default 10 128 3 0 est le nom de la file d'attente. La file d'attente poussée ci-dessus a un default et doit être remplacée par email .emailAprès avoir lancé la commande, regardons le code : exécutez d'abord : la méthode
ControllerWorkerController, nous suivons le code dans la méthode actionListen, où il boucle en fait tout le temps pour exécuter les tâches de la file d'attente des opérations : vendorshmilyzxtqueueWorker.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();
                }
               }
    }

Remarque :

est une méthode $queue->pop($queueName); utilisant des transactions pour exécuter SQL et créer une instance de vendorshmilyzxtqueuequeuesDatabaseQueue.php vendorshmilyzxtqueuejobsDatabaseJob.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;
    }

Quant à :

est exécuté par $job->execute(); héritant de la classe parent DatabaseJob Suivez le code pour trouver l'événement exécuté par Job, yiibaseComponent trigger

/** 
 * 执行任务 
 */
public function execute(){   
  //beforeExecute 执行任务之前的一个事件 在JobEvent中并没有什么可执行的代码
  $this->trigger(self::EVENT_BEFORE_EXECUTE, new JobEvent(["job" => $this, 'payload' => $this->getPayload()]));
  $this->resolveAndFire();//真正执行的任务的方法
}

Enfin nous arrivons à la

 /**   * 真正任务执行方法（调用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();
        }
    }

classe

exécutée, voici les objets et les données poussés vers la file d'attente, puis notre logique de traitement.SendMailhandle($job,$data)

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!