Yii2 佇列 shmilyzxt/yii2-queue 簡單概述

NO IMAGE

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'=>'[email protected]','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 '<pre>';print_r($payload);
//$payload即任務的資料,你拿到任務資料後就可以執行發郵件了
//TODO 發郵件
}

總結

以上所述是小編給大家介紹的Yii2 佇列 shmilyzxt/yii2-queue簡介,希望對大家有所幫助,如果大家有任何疑問請給我留言,小編會及時回覆大家的。在此也非常感謝大家對指令碼之家網站的支援!