1.我用的yii2高級版,我們從配置開始看代碼,這里我用的是mysql隊列,首先配置文件,我把queue配置項寫在根目錄common/config/main-local.php下的 components數組下,更改一下數據庫配置.復制composer安裝后復制
vendor/shmilyzxt/yii2-queue/jobs/jobs.sqlvendor/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 '<pre>';print_r($payload); //$payload即任務的數據，你拿到任務數據后就可以執行發郵件了 //TODO 發郵件 }總結
以上所述是小編給大家介紹的Yii2 隊列 shmilyzxt/yii2-queue簡介，希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時回復大家的。在此也非常感謝大家對VeVb武林網網站的支持！
新聞熱點
疑難解答
圖片精選
