queue:work 原理分析

Epona 的个人博客 / 10 / 2 / 创建于 6年前 / 更新于 6年前

最近在项目中使用了队列，因此研究一下相关的源码。本文只是粗略的进行分析，如果发现了错误欢迎大家讨论交流。

基本实现

php artisan queue:work 的代码实现是在Illuminate\Queue\Console\WorkCommand中。那么，让我们看一下它是怎样处理的：

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);  
}

protected function runWorker($connection, $queue)
{
    // 设置缓存
    $this->worker->setCache($this->laravel['cache']->driver());

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

上面的代码逻辑很简单，代码的核心逻辑在最后一行runWorker中，根据传参的不同，执行方法可能为 runNextJob 或者 daemon，由于本文的目的为研究原理，故只分析 daemon 方法。

daemon方法

daemon 方法存在于Illuminate\Queue\Worker中，具体如下：

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

    $lastRestart = $this->getTimestampOfLastQueueRestart();

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

            continue;
        }

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

        $this->registerTimeoutHandler($job, $options);

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

        $this->stopIfNecessary($options, $lastRestart);
    }
}

下面让我们一步一步来分析其中的代码逻辑。

监听信号

protected function listenForSignals()
{
    if ($this->supportsAsyncSignals()) {
        pcntl_async_signals(true);

        pcntl_signal(SIGTERM, function () {
            $this->shouldQuit = true;
        });

        pcntl_signal(SIGUSR2, function () {
            $this->paused = true;
        });

        pcntl_signal(SIGCONT, function () {
            $this->paused = false;
        });
    }
}

这段代码主要用到了 PHP 的 pcntl 扩展，具体的介绍和使用方法请看官方文档以及这里。

此扩展的主要用于进程控制，根据相关的进程信号值设置对应的参数。

获取上一次重启时间

   protected function getTimestampOfLastQueueRestart()                                                                                    
{                                                                                                                                      
     if ($this->cache) {                                                                                                                
         return $this->cache->get('illuminate:queue:restart');                                                                          
     }                                                                                                                                  
}

当上一步之行完之后，应用会获取上一次的队列重启时间戳，即运行命令 php artisan queue:restart 时的时间戳。此时，如果缓存使用Redis的话，使用 redis-cli 登录后并输入 monitor 然后执行的话（monitor 命令用于实时监控 redis 的操作），会看到如下的显示，

1434697488.632958 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697491.634111 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697494.635239 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697497.636391 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697500.637753 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697503.639073 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697506.640155 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697509.641288 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"
1434697512.642365 [0 127.0.0.1:60136] "GET" "laravel:illuminate:queue:restart"

这表示我们正在不断的获取时间戳。

判断是否需要执行队列

接着在一个无限的循环中，我们判断是否可以执行队列，如果答案为否，那么我们将其进行休眠（ PHP 中的 sleep 方法），如有必要也可以将对应的进程杀掉（stopIfNecessary 方法）。

执行队列

如果队列可以执行，那么代码将会执行对应的队列。

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);
    }
}

public function process($connectionName, $job, WorkerOptions $options)
{
    $this->raiseBeforeJobEvent($connectionName, $job);

    $this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
        $connectionName, $job, (int) $options->maxTries
    );

    $job->fire();

    $this->raiseAfterJobEvent($connectionName, $job);

    // 省略了错误处理部分的代码
}

我们可以看到runJob的逻辑很简单，代码进行队列处理（process 方法），如果碰到错误或者异常将抛出对应的异常。在 process 方法中首先触发事件，然后判断是否达到最大尝试次数。如果没有，则开始队列处理，fire 方法会根据我们使用那种队列（Redis，Beanstalkd等）来进行对应的实现；

队列执行

// Illuminate\Queue\Jobs\Job;
public function fire()
{
    $payload = $this->payload();

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

    ($this->instance = $this->resolve($class))->{$method}{$this, $payload['dada']};
}

在这里我们通过 payload来获取要执行的对象和方法，其中 payload 内容如下：

[2019-05-16 11:02:04] local.INFO: array (
  'displayName' => 'App\\Jobs\\TestJob',
  'job' => 'Illuminate\\Queue\\CallQueuedHandler@call',
  'maxTries' => NULL,
  'timeout' => NULL,
  'timeoutAt' => NULL,
  'data' => 
  array (
    'commandName' => 'App\\Jobs\\TestJob',
    'command' => 'O:16:"App\\Jobs\\TestJob":7:{s:6:"' . "\0" . '*' . "\0" . 'job";N;s:10:"connection";N;s:5:"queue";N;s:15:"chainConnection";N;s:10:"chainQueue";N;s:5:"delay";N;s:7:"chained";a:0:{}}',
  ),
)

因此，fire 最后会使用 Illuminate\Queue\CallQueueHandler类中的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)
    );

    // 其余错误处理代码省略
}

在这里我们首先设定好 Job 实例，然后处理对应的命令和错误，至此，queue:work 的基本原理已经分析完毕。

queue