I feel very good during the use. I suggest you take a look at the principles of queues.
##shmilyzxt/yii2-queue Simple explanation:
queueThe configuration items are written under the
components array in the root directory
common\config\main-local.php, change the database configuration. Copy
composerCopy after installation
vendor\shmilyzxt\yii2-queue\jobs\jobs.sql##vendor\shmilyzxt\yii2-queue\failed\failed.sql
2 sql files to the database to create a queue data table and a data table when the task fails.
Let’s go to vendor\shmilyzxt\queue\queues\DatabaseQueue.php
to take a look at the code. The pushOn()
method is written in DatabaseQueue
Parent class of classvendor\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}"); } }
Comment: Here is the most Let’s take a look at the yii2 event event class.
$this->push($job, $data, $queue);
, here is the cooperation withqueue View class files, jump to related functions, process the data and record it in the database. (Function direction:
getQueue()-->createPayload()-->pushToDatabase()),
pushOn()Finally returns the result of data insertion into the database. Successful
$ret is 1.
3. Run the command processing queue in the background, for example:
php ./ yii worker/listen default 10 128 3 0
default is the name of the queue. The
email queue pushed above should be changed to
email.
After starting the command, let's look at the code: first execute: WorkerController
Controller
actionListen method, we follow the code and enter
vendor\shmilyzxt\queue\Worker.php -- In the listen method, it is actually looping all the time to execute the tasks of the operation queue:
##
/** * 启用一个队列后台监听任务 * @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);
is
uses transactions to execute SQL within the method, and creates vendor\shmilyzxt\queue\jobs\DatabaseJob.php
The instance of ##<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false;"> //取出一个任务
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, [
&#39;class&#39; => &#39;shmilyzxt\queue\jobs\DatabaseJob&#39;,
&#39;queue&#39; => $queue,
&#39;job&#39; => $job,
&#39;queueInstance&#39; => $this,
]);
return \Yii::createObject($config);
} $tran->commit();
return false;
}</pre><div class="contentsignin">Copy after login</div></div>
as for:
is
DatabaseJob
Job Executed, follow the code to find the event executed by
yii\base\Component trigger,
<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false;">/**
* 执行任务
*/
public function execute(){
//beforeExecute 执行任务之前的一个事件 在JobEvent中并没有什么可执行的代码
$this->trigger(self::EVENT_BEFORE_EXECUTE, new JobEvent(["job" => $this, &#39;payload&#39; => $this->getPayload()]));
$this->resolveAndFire();//真正执行的任务的方法
}</pre><div class="contentsignin">Copy after login</div></div>
/** * 真正任务执行方法(调用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(); } }
of the executed
SendMail
The above is the detailed content of Introduction to PHPyii2 queue shmilyzxt/yii2-queue. For more information, please follow other related articles on the PHP Chinese website!public function handle($job,$data)
{ if($job->getAttempts() > 3){ $this->failed($job);
} $payload = $job->getPayload();
echo '<pre class="brush:php;toolbar:false">';print_r($payload);
//$payload即任务的数据,你拿到任务数据后就可以执行发邮件了
//TODO 发邮件
}