【Laravel-海賊王系列】第十一章，Job&隊列消費端實現

時間 2019-11-16

標籤 Laravel-海賊王系列第十一章 job 隊列消費實現简体版

原文原文鏈接

啓動指令

php artisan queue:workphp

啓動文件

namespace Illuminate\Queue\Console;

use Illuminate\Queue\Worker;
use Illuminate\Support\Carbon;
use Illuminate\Console\Command;
use Illuminate\Contracts\Queue\Job;
use Illuminate\Queue\WorkerOptions;
use Illuminate\Queue\Events\JobFailed;
use Illuminate\Queue\Events\JobProcessed;
use Illuminate\Queue\Events\JobProcessing;

class WorkCommand extends Command
{
  
    ...
    
    /**
     * @var \Illuminate\Queue\Worker
     */
    protected $worker;

    public function __construct(Worker $worker)
    {
        parent::__construct();
        $this->worker = $worker;

    }
    
    public function handle()
    {

        if ($this->downForMaintenance() && $this->option('once')) {
            return $this->worker->sleep($this->option('sleep'));
        }

        $this->listenForEvents();

        $connection = $this->argument('connection')
                        ?: $this->laravel['config']['queue.default'];

        $queue = $this->getQueue($connection);

        $this->runWorker(
            $connection, $queue
        );
    }

   ...
}

複製代碼

咱們先從構造函數和 handle() 方法開始分析，這是入口。html

片斷一：判斷是否維護模式或者 --force 強制啓動laravel

if ($this->downForMaintenance() && $this->option('once')) {
    return $this->worker->sleep($this->option('sleep'));
}
複製代碼

片斷二：經過事件綁定在控制檯輸出信息redis

$this->listenForEvents();

protected function listenForEvents()
{
    $this->laravel['events']->listen(JobProcessing::class, function ($event) {
        $this->writeOutput($event->job, 'starting');
    });

    $this->laravel['events']->listen(JobProcessed::class, function ($event) {
        $this->writeOutput($event->job, 'success');
    });

    $this->laravel['events']->listen(JobFailed::class, function ($event) {
        $this->writeOutput($event->job, 'failed');

        $this->logFailedJob($event);
    });
}
複製代碼

片斷三：經過配置文件中配置的驅動獲取對應驅動的隊列名,若是沒有則返回 defaultjson

$connection = $this->argument('connection') ?: $this->laravel['config']['queue.default'];

$queue = $this->getQueue($connection);

protected function getQueue($connection)
{
    return $this->option('queue') ?: $this->laravel['config']->get(
        "queue.connections.{$connection}.queue", 'default'
    );
}
    
複製代碼

片斷四：傳入鏈接驅動和隊列名稱到 runWorker 方法運行任務。bash

$this->runWorker(
            $connection, $queue
        );
複製代碼

這裏是啓動的重點，咱們傳入的 $connection = 'redis' $queue = 'default'，繼續分析app

protected function runWorker($connection, $queue)
{
    // "這裏的 $this->laravel['cache'] 是 Illuminate\Cache\CacheManager 類的實例。 （是在 app.providers.Illuminate\Cache\CacheServiceProvider::class 註冊的） $this->laravel['cache']->driver() 返回 Illuminate\Cache\Repository 類的實例。"

    // "框架經過 CacheManager 對不少存儲管理進行了統一。 能夠經過修改 app.config.cache.default 和 `app.config.cache.stores 中的值來修改存儲驅動。"
    
    // "將獲取的驅動賦值給 workder 的 cache成員"
    $this->worker->setCache($this->laravel['cache']->driver());
    
    // "當 worker 對象擁有了cache對象以後便擁有了操做對應數據的能力 ！"
    return $this->worker->{$this->option('once') ? 'runNextJob' : 'daemon'}(
        $connection, $queue, $this->gatherWorkerOptions()
    );
}
複製代碼

繼續運行框架

return $this->worker->{$this->option('once') ? 'runNextJob' : 'daemon'}(
        $connection, $queue, $this->gatherWorkerOptions()
    );
複製代碼

這裏傳入的參數分別是，能夠看出都是對隊列消費的一些基本設置。異步

當運行模式非 --once 的狀況下就會以 daemon 的方式運行。async

咱們看 \Illuminate\Queue\Worker 對象的 daemon 方法便可

守護進程模式

public function daemon($connectionName, $queue, WorkerOptions $options)
{
    if ($this->supportsAsyncSignals()) {
        $this->listenForSignals();
    }

    $lastRestart = $this->getTimestampOfLastQueueRestart();

    while (true) {
    
        if (! $this->daemonShouldRun($options, $connectionName, $queue)) {
            $this->pauseWorker($options, $lastRestart);

            continue;
        }

        $job = $this->getNextJob(
            $this->manager->connection($connectionName), $queue
        );

        if ($this->supportsAsyncSignals()) {
            $this->registerTimeoutHandler($job, $options);
        }

        if ($job) {
            $this->runJob($job, $connectionName, $options);
        } else {
            $this->sleep($options->sleep);
        }

        $this->stopIfNecessary($options, $lastRestart, $job);
    }
}
複製代碼

進程參數設定

先設置進程的一些管理參數

if ($this->supportsAsyncSignals()) { // extension_loaded('pcntl'); 是否支持 'pcntl' 拓展，支持多進程的拓展。
        $this->listenForSignals();
}

protected function listenForSignals()
{
    // "PHP7.1信號新特性 -- 開啓異步信號處理"
    pcntl_async_signals(true);

    // "安裝信號處理器，後面能夠傳入相應的信號來終止或其餘操做"
    pcntl_signal(SIGTERM, function () {
        // "SIGTERM 終止進程 軟件終止信號"
        $this->shouldQuit = true; 
    });

    pcntl_signal(SIGUSR2, function () {
        // "SIGUSR2 終止進程 用戶定義信號2"
        $this->paused = true;
    });

    pcntl_signal(SIGCONT, function () {
        // "SIGCONT 忽略信號 繼續執行一箇中止的進程"
        $this->paused = false;
    });
}
複製代碼

關於 pcntl 的用法能夠參考 PCNTL

信號能夠參考對照表

接着看，從 cache 中獲取上一次重啓的時間戳

$lastRestart = $this->getTimestampOfLastQueueRestart();
複製代碼

循環任務執行

判斷是否終止運行

if (! $this->daemonShouldRun($options, $connectionName,$queue)) {
    
    // "$opions 就是 調用artisan 傳入的參數 $connectionName 我用了redis驅動，全部就是 'redis' $queue 這裏沒有傳入隊列則是 'default'"
    
    $this->pauseWorker($options, $lastRestart);
    continue;
}
複製代碼

下面代碼一共三個判斷：

1.是不是關站模式而且非強制運行。

2.是否有外部傳入的暫停信號

3.是否有綁定 Looping 事件執行並返回結果

若是符合條件則暫停或者發送終止信號。

主要功能是爲了控制是否繼續執行任務。

protected function daemonShouldRun(WorkerOptions $options, $connectionName, $queue)
{
    return ! (($this->manager->isDownForMaintenance() && ! $options->force) ||
        $this->paused ||
        $this->events->until(new Events\Looping($connectionName, $queue)) === false);
}
複製代碼

獲取待運行的 `Job`

// "$this->manager->connection($connectionName) 是 Illuminate\Queue\RedisQueue 對象 $queue ： 'default'"

$job = $this->getNextJob(
    $this->manager->connection($connectionName), $queue
);
複製代碼

繼續看 getNextJob

protected function getNextJob($connection, $queue)
{
    try {
        foreach (explode(',', $queue) as $queue) {
            if (! is_null($job = $connection->pop($queue))) {
                return $job;
            }
        }
    } catch (Exception $e) {
    // "異常處理主要是報告異常"
    // "設置 '$this->shouldQuit = true;' 後續就會終止"
        $this->exceptions->report($e);

        $this->stopWorkerIfLostConnection($e);

        $this->sleep(1);
    } catch (Throwable $e) {
        $this->exceptions->report($e = new FatalThrowableError($e));

        $this->stopWorkerIfLostConnection($e);

        $this->sleep(1);
    }
}
複製代碼

上面分析過了 $connection 是 RedisQueue 對象，全部展開 RedisQueue 的 pop 方法，獲取要執行的任務對象。

public function pop($queue = null)
{
    $this->migrate($prefixed = $this->getQueue($queue));

    if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
        return;
    }
    [$job, $reserved] = $nextJob;
    if ($reserved) {
        return new RedisJob(
            $this->container, $this, $job,
            $reserved, $this->connectionName, $queue ?: $this->default
        );
    }
}
複製代碼

遷移延遲隊列

在 pop 的過程當中首先遷移延遲隊列的相關數據

protected function migrate($queue)
{
    // "這裏是否是很熟悉了，上一章存儲端分析的時候延遲"
    // "隊列就是用的這個key來存的"
    
    // "將延遲的隊列遷移到主隊列"
    $this->migrateExpiredJobs($queue.':delayed', $queue);
    
    // "將過時隊列遷移到主隊列"
    if (! is_null($this->retryAfter)) {
        $this->migrateExpiredJobs($queue.':reserved', $queue);
    }
}
複製代碼

繼續看如何遷移到主隊列的

public function migrateExpiredJobs($from, $to)
{
    return $this->getConnection()->eval(
        LuaScripts::migrateExpiredJobs(), 
        2,
        $from,
        $to,
        $this->currentTime()
    );
}

public static function migrateExpiredJobs()
{
    return <<<'LUA'
    if(next(val) ~= nil) then
        redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)
        
        for i = 1, #val, 100 do
            redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
        end
    end
    
    return val
    LUA;
}    
複製代碼

最終經過 eval 命令使用 Lua 解釋器執行腳本。請看 Redis Eval

真香，這僅僅是把延遲任務切回主隊列，繼續！

檢索數據

從隊列檢索下一個 Job

if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
            return; // 沒有數據就返回
        }
複製代碼

展開檢索代碼

protected function retrieveNextJob($queue)
{
    // "默認值是 null"
    if (! is_null($this->blockFor)) {
        return $this->blockingPop($queue);
    }

    // "這段是直接經過 lua 從 redis lpop出對象,"
    // "在lua中完成封裝,執行邏輯和 blockingPop 類似"
    return $this->getConnection()->eval(
        LuaScripts::pop(), 2, $queue, $queue.':reserved',
        $this->availableAt($this->retryAfter)
    );
}
複製代碼

咱們主要看 blockingPop 的代碼

protected function blockingPop($queue)
{
    // "以阻塞的方式彈出隊列的第一個元素"
    $rawBody = $this->getConnection()->blpop($queue, $this->blockFor);
    
    // "解析獲取的數據，同時再封裝一個重試對象並寫入有序集合。"
    if (! empty($rawBody)) {
        $payload = json_decode($rawBody[1], true);

        $payload['attempts']++;

        $reserved = json_encode($payload);

        $this->getConnection()->zadd($queue.':reserved', [
            $reserved => $this->availableAt($this->retryAfter),
        ]);

        return [$rawBody[1], $reserved];
    }

    return [null, null];
}
複製代碼

檢索完成以後回到 pop 中繼續執行

public function pop($queue = null)
{
    $this->migrate($prefixed = $this->getQueue($queue));

    if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
        return;
    }
    
    // "到這裏了！"
    [$job, $reserved] = $nextJob;
    if ($reserved) {
        return new RedisJob(
            $this->container, $this, $job,
            $reserved, $this->connectionName, $queue ?: $this->default
        );
    }
}
複製代碼

咱們來看看 $nextJob 是什麼

最後調用

return new RedisJob(
                $this->container, $this, $job,
                $reserved, $this->connectionName, $queue ?: $this->default
            );
複製代碼

看看 Illuminate\Queue\Jobs\RedisJob 的構造函數

public function __construct(Container $container, RedisQueue $redis, $job, $reserved, $connectionName, $queue)
{
    $this->job = $job;
    $this->redis = $redis;
    $this->queue = $queue;
    $this->reserved = $reserved;
    $this->container = $container;
    $this->connectionName = $connectionName;

    $this->decoded = $this->payload();
}
複製代碼

這應該是最後一層封裝,最後要返回給最外層的任務對象。

運行 `Job`

回到 Worker 對象中

...

$job = $this->getNextJob(
            $this->manager->connection($connectionName), $queue
        );

// "剛剛咱們從 redis 中拿到了封裝好的 $job 對象，繼續執行"

// "$job 就是 Illuminate\Queue\Jobs\RedisJob 對象"

// "是否支持 pcntl 拓展，異步模式傳遞信號"
if ($this->supportsAsyncSignals()) {
// "設置超時信號處理"
$this->registerTimeoutHandler($job, $options);
}
複製代碼

繼續註冊超時信號控制

protected function registerTimeoutHandler($job, WorkerOptions $options)
{
    pcntl_signal(SIGALRM, function () {
        $this->kill(1);
    });

    pcntl_alarm(
        max($this->timeoutForJob($job, $options), 0)
    );
}
複製代碼

總算要到運行 Job 的部分了

if ($job) {
        $this->runJob($job, $connectionName, $options);
    } else {
        // "不存在 $job 則睡眠,最低睡眠1秒"
        $this->sleep($options->sleep);
    }
複製代碼

解析 `runJob`

到這一步咱們已經拿到了全部的對象,接下來就是把對象用起來！

protected function runJob($job, $connectionName, WorkerOptions $options)
{
    try {
        return $this->process($connectionName, $job, $options);
    } catch (Exception $e) {
        
        // "異常處理和上部分的同樣，"
        // "設定中止信號，在循環的結尾會檢測信號"
        // "所以咱們不須要分析這段"
        
        $this->exceptions->report($e);

        $this->stopWorkerIfLostConnection($e);
    } catch (Throwable $e) {
        $this->exceptions->report($e = new FatalThrowableError($e));

        $this->stopWorkerIfLostConnection($e);
    }
}
複製代碼

展開

$this->process($connectionName, $job, $options);
複製代碼

繼續展開

public function process($connectionName, $job, WorkerOptions $options)
{
    try {
       
        // "觸發任務執行前的綁定事件，從隊列刪除任務"
        $this->raiseBeforeJobEvent($connectionName, $job);
      
        // "標記超過最大重試次數的任務"
        $this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
            $connectionName, $job, (int) $options->maxTries
        );

       
        $job->fire();

        // "觸發任務執行後的綁定事件"
        $this->raiseAfterJobEvent($connectionName, $job);
    } catch (Exception $e) {
        $this->handleJobException($connectionName, $job, $options, $e);
    } catch (Throwable $e) {
        $this->handleJobException(
            $connectionName, $job, $options, new FatalThrowableError($e)
        );
    }
}
複製代碼

`$job->fire()`

$job => Illuminate\Queue\Jobs\RedisJob 繼承了 Illuminate\Queue\Jobs\Job 因此調用了抽象父類的 fire() 方法

public function fire()
{
    $payload = $this->payload();

    [$class, $method] = JobName::parse($payload['job']);

    ($this->instance = $this->resolve($class))->{$method}($this, $payload['data']);
}
複製代碼

咱們看看 $payload 的結構實際就是 json_decode($job, true)

轉換後的[$class, $method] 分別是 Illuminate\Queue\CallQueuedHandler 和 call

最後就是從容器中解析出 Illuminate\Queue\CallQueuedHandler 對象而且調用 call 方法，展開方法

public function call(Job $job, array $data)
{
    try {
        $command = $this->setJobInstanceIfNecessary(
            $job, unserialize($data['command'])
        );
    } catch (ModelNotFoundException $e) {
        return $this->handleModelNotFound($job, $e);
    }

    $this->dispatcher->dispatchNow(
        $command, $this->resolveHandler($job, $command)
    ); 

    if (! $job->hasFailed() && ! $job->isReleased()) {
        $this->ensureNextJobInChainIsDispatched($command);
    }

    if (! $job->isDeletedOrReleased()) {
        $job->delete();
    }
}
複製代碼

先看看 $command 獲取的是什麼

protected function setJobInstanceIfNecessary(Job $job, $instance)
{
    if (in_array(InteractsWithQueue::class, class_uses_recursive($instance))) {
        $instance->setJob($job);
    }

    return $instance;
}
複製代碼

打印 class_uses_recursive($instance)

接着就調用了 $instance->setJob($job);

這裏的 $instance 就是對應咱們本身編寫的任務對象。

執行完以後最終 $command 返回的就是本身編寫的類

將 RedisJob 和 $command 傳給 dispatchNow 方法 $this->dispatcher 是 Illuminate\Bus\Dispatcher 對象

$this->dispatcher->dispatchNow(
            $command, $this->resolveHandler($job, $command)
        );
複製代碼

最後的真像

public function dispatchNow($command, $handler = null)
    {
        if ($handler || $handler = $this->getCommandHandler($command)) {
            $callback = function ($command) use ($handler) {
                // "劃重點，要考!"
                return $handler->handle($command); 
            };
        } else {
            $callback = function ($command) {
                return $this->container->call([$command, 'handle']);
            };
        }

        return $this->pipeline->send($command)->through($this->pipes)->then($callback);
    }
複製代碼