shmilyzxt/yii2-queue 簡(jiǎn)單解釋:
1.我用的yii2高級(jí)版,我們從配置開(kāi)始看代碼,這里我用的是mysql隊(duì)列,首先配置文件,我把queue配置項(xiàng)寫在根目錄common\config\main-local.php下的 components數(shù)組下,更改一下數(shù)據(jù)庫(kù)配置.復(fù)制composer安裝后復(fù)制
vendor\shmilyzxt\yii2-queue\jobs\jobs.sql vendor\shmilyzxt\yii2-queue\failed\failed.sql
2個(gè)sql文件到數(shù)據(jù)庫(kù)中建立隊(duì)列數(shù)據(jù)表和執(zhí)行任務(wù)失敗時(shí)的數(shù)據(jù)表.
2.推送任務(wù)開(kāi)始語(yǔ)法:\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中:
//入隊(duì)列
public function pushOn($job, $data = '', $queue = null)
{
//canPush 檢查隊(duì)列是否已達(dá)最大任務(wù)量
if ($this->canPush()) {
//beforePush 入隊(duì)列前的事件
$this->trigger(self::EVENT_BEFORE_PUSH);
//入隊(duì)列
$ret = $this->push($job, $data, $queue);
//afterPush 入隊(duì)列后的事件
$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
關(guān)于入隊(duì)列: $this->push($job, $data, $queue);,這里在配合queue類文件查看,相關(guān)函數(shù)跳轉(zhuǎn),處理一下數(shù)據(jù)記錄到數(shù)據(jù)庫(kù)中.(函數(shù)走向:getQueue()-->createPayload()-->pushToDatabase()),pushOn()最終返回?cái)?shù)據(jù)插入數(shù)據(jù)庫(kù)的結(jié)果,成功$ret是1.
3.后臺(tái)運(yùn)行命令處理隊(duì)列,例:php ./yii worker/listen default 10 128 3 0 其中default是隊(duì)列的名稱,上面推送了一個(gè)email隊(duì)列 應(yīng)該改為email.
啟動(dòng)命令后,我們來(lái)看代碼:首先執(zhí)行:WorkerController控制器 actionListen方法,我們跟著代碼進(jìn)入到 vendor\shmilyzxt\queue\Worker.php -- listen方法中,這里其實(shí)就是一直在循環(huán),執(zhí)行操作隊(duì)列的任務(wù):
/**
* 啟用一個(gè)隊(duì)列后臺(tái)監(jiān)聽(tīng)任務(wù)
* @param Queue $queue
* @param string $queueName 監(jiān)聽(tīng)隊(duì)列的名稱(在pushon的時(shí)候把任務(wù)推送到哪個(gè)隊(duì)列，則需要監(jiān)聽(tīng)相應(yīng)的隊(duì)列才能獲取任務(wù))
* @param int $attempt 隊(duì)列任務(wù)失敗嘗試次數(shù)，0為不限制
* @param int $memory 允許使用的最大內(nèi)存
* @param int $sleep 每次檢測(cè)的時(shí)間間隔
*/
public static function listen(Queue $queue, $queueName = 'default', $attempt = 10, $memory = 512, $sleep = 3, $delay = 0){
while (true){
try{
//DatabaseQueue從數(shù)據(jù)庫(kù)隊(duì)列取出一個(gè)可用任務(wù)(實(shí)例),并且更新任務(wù)
$job = $queue->pop($queueName);
}catch (\Exception $e){
throw $e;
continue;
}
if($job instanceof Job){
//判斷執(zhí)行錯(cuò)誤的次數(shù)是否大于傳入的執(zhí)行次數(shù)
if($attempt > 0 && $job->getAttempts() > $attempt){
$job->failed();
}else{
try{
//throw new \Exception("test failed");
$job->execute();
}catch (\Exception $e){
//執(zhí)行失敗,判斷是否被刪除,重新入隊(duì)
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方法內(nèi)使用事務(wù)執(zhí)行SQL,并且創(chuàng)建vendor\shmilyzxt\queue\jobs\DatabaseJob.php的實(shí)例
//取出一個(gè)任務(wù)
public function pop($queue = null)
{
$queue = $this->getQueue($queue);
if (!is_null($this->expire)) {
//$this->releaseJobsThatHaveBeenReservedTooLong($queue);
}
$tran = $this->connector->beginTransaction();
//判斷是否有一個(gè)可用的任務(wù)需要執(zhí)行
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執(zhí)行的,順著代碼找下去是yii\base\Component trigger執(zhí)行的事件,
/**
* 執(zhí)行任務(wù)
*/
public function execute()
{
$this->trigger(self::EVENT_BEFORE_EXECUTE, new JobEvent(["job" => $this, 'payload' => $this->getPayload()]));//beforeExecute 執(zhí)行任務(wù)之前的一個(gè)事件 在JobEvent中并沒(méi)有什么可執(zhí)行的代碼
$this->resolveAndFire();//真正執(zhí)行的任務(wù)的方法
}
/**
* 真正任務(wù)執(zhí)行方法（調(diào)用hander的handle方法）
* @param array $payload
* @return void
*/
protected function resolveAndFire()
{
$payload = $this->getPayload();
$payload = unserialize($payload); //反序列化數(shù)據(jù)
$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 {//執(zhí)行的`SendMail`類的`handle($job,$data)`方法
$this->handler = $this->getHander($class);
$this->handler->handle($this, $payload['data']);
}
//執(zhí)行完任務(wù)后刪除
if (!$this->isDeletedOrReleased()) {
$this->delete();
}
}