登录
当前位置： 首页 > 工具软件 > laravel-failed-job-monitor > 使用案例 >

Laravel Queue——消息队列任务处理器源码剖析

柳奇思
2023-12-01

运行队列处理器

队列处理器的设置

Laravel 包含一个队列处理器，当新任务被推到队列中时它能处理这些任务。你可以通过 queue:work 命令来运行处理器。要注意，一旦 queue:work 命令开始，它将一直运行，直到你手动停止或者你关闭控制台：

php artisan queue:work

  • 可以指定队列处理器所使用的连接。 php artisan queue:work redis
  • 可以自定义队列处理器，方式是处理给定连接的特定队列。 php artisan queue:work redis --queue=emails
  • 可以使用 --once 选项来指定仅对队列中的单一任务进行处理： php artisan queue:work --once
  • 如果一个任务失败了，会被放入延时队列中取，–delay 选项可以设置失败任务的延时时间： php artisan queue:work --delay=2
  • 如果想要限制一个任务的内存，可以使用 --memory: php artisan queue:work --memory=128
  • 当队列需要处理任务时，进程将继续处理任务，它们之间没有延迟。但是，如果没有新的工作可用，–sleep 参数决定了工作进程将 「睡眠」 多长时间： php artisan queue:work --sleep=3
  • 可以指定 Laravel 队列处理器最多执行多长时间后就应该被关闭掉： php artisan queue:work --timeout=60
  • 可以指定 Laravel 队列处理器失败任务重试的次数： php artisan queue:work --tries=60
  • 可以看出来，队列处理器的设置大多数都可以由任务类进行设置，但是其中三个 sleep、delay、memory 只能由 artisan 来设置。

WorkCommand 命令行启动

任务处理器进程的命令行模式会调用 Illuminate\Queue\Console\WorkCommand，这个类在初始化的时候依赖注入了 Illuminate\Queue\Worker:

class WorkCommand extends Command
{
    protected $signature = 'queue:work
                            {connection? : The name of connection}
                            {--queue= : The queue to listen on}
                            {--daemon : Run the worker in daemon mode (Deprecated)}
                            {--once : Only process the next job on the queue}
                            {--delay=0 : Amount of time to delay failed jobs}
                            {--force : Force the worker to run even in maintenance mode}
                            {--memory=128 : The memory limit in megabytes}
                            {--sleep=3 : Number of seconds to sleep when no job is available}
                            {--timeout=60 : The number of seconds a child process can run}
                            {--tries=0 : Number of times to attempt a job before logging it failed}';

    public function __construct(Worker $worker)
    {
        parent::__construct();

        $this->worker = $worker;
    }                         

    public function fire()
    {
        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
        );
    }
}

任务处理器启动后，会运行 fire 函数，在执行任务之前，程序首先会注册监听事件，主要监听任务完成与任务失败的情况：

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

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

        $this->logFailedJob($event);
    });
}

protected function writeOutput(Job $job, $failed)
{
    if ($failed) {
        $this->output->writeln('<error>['.Carbon::now()->format('Y-m-d H:i:s').'] Failed:</error> '.$job->resolveName());
    } else {
        $this->output->writeln('<info>['.Carbon::now()->format('Y-m-d H:i:s').'] Processed:</info> '.$job->resolveName());
    }
}

protected function logFailedJob(JobFailed $event)
{
    $this->laravel['queue.failer']->log(
        $event->connectionName, $event->job->getQueue(),
        $event->job->getRawBody(), $event->exception
    );
}

启动任务管理器 runWorker, 该函数默认会调用 Illuminate\Queue\Worker 的 daemon 函数，只有在命令中强制 --once 参数的时候，才会执行 runNestJob 函数:

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

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

Worker 任务调度

我们接下来接着看 daemon 函数：

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

    $lastRestart = $this->getTimestampOfLastQueueRestart();

    while (true) {
        if (! $this->daemonShouldRun($options)) {
            $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);
    }
}
信号处理

listenForSignals 函数用于 PHP 7.1 版本以上，用于脚本的信号处理。所谓的信号处理，就是由 Process Monitor（如 Supervisor ）发送并与我们的脚本进行通信的异步通知。

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

protected function supportsAsyncSignals()
{
    return version_compare(PHP_VERSION, '7.1.0') >= 0 &&
           extension_loaded('pcntl');
}

pcntl_async_signals() 被调用来启用信号处理，然后我们为多个信号注册处理程序：

  • 当脚本被 Supervisor 指示关闭时，会引发信号 SIGTERM
  • SIGUSR2 是用户定义的信号，Laravel 用来表示脚本应该暂停。
  • 当暂停的脚本被 Supervisor 指示继续进行时，会引发 SIGCONT

在真正运行任务之前，程序还从 cache 中取了一次最后一次重启的时间：

protected function getTimestampOfLastQueueRestart()
{
    if ($this->cache) {
        return $this->cache->get('illuminate:queue:restart');
    }
}
确定 worker 是否应该处理作业

进入循环后，首先要判断当前脚本是应该处理任务，还是应该暂停，还是应该退出：

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

以下几种情况，循环将不会处理任务：

  • 脚本处于 维护模式 并且没有 --force 选项
  • 脚本被 supervisor 暂停
  • 脚本的 looping 事件监听器返回 false

looping 事件监听器在每次循环的时候都会被启动，如果返回 false，那么当前的循环将会被暂停：pauseWorker:

protected function pauseWorker(WorkerOptions $options, $lastRestart)
{
    $this->sleep($options->sleep > 0 ? $options->sleep : 1);

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

脚本在 sleep 一段时间之后，就要重新判断当前脚本是否需要 stop：

protected function stopIfNecessary(WorkerOptions $options, $lastRestart)
{
    if ($this->shouldQuit) {
        $this->kill();
    }

    if ($this->memoryExceeded($options->memory)) {
        $this->stop(12);
    } elseif ($this->queueShouldRestart($lastRestart)) {
        $this->stop();
    }
}

protected function queueShouldRestart($lastRestart)
{
    return $this->getTimestampOfLastQueueRestart() != $lastRestart;
}

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

以下情况脚本将会被 stop：

  • 脚本被 supervisor 退出
  • 内存超限
  • 脚本被重启过
public function kill($status = 0)
{
    if (extension_loaded('posix')) {
        posix_kill(getmypid(), SIGKILL);
    }

    exit($status);
}

public function stop($status = 0)
{
    $this->events->fire(new Events\WorkerStopping);

    exit($status);
}

脚本被重启，当前的进程需要退出并且重新加载。

获取下一个任务

当含有多个队列的时候，命令行可以用 , 连接多个队列的名字，位于前面的队列优先级更高：

protected function getNextJob($connection, $queue)
{
    try {
        foreach (explode(',', $queue) as $queue) {
            if (! is_null($job = $connection->pop($queue))) {
                return $job;
            }
        }
    } catch (Exception $e) {
        $this->exceptions->report($e);
    } catch (Throwable $e) {
        $this->exceptions->report(new FatalThrowableError($e));
    }
}

$connection 是具体的驱动，我们这里是 Illuminate\Queue\RedisQueue:

class RedisQueue extends Queue implements QueueContract
{
    public function pop($queue = null)
    {
        $this->migrate($prefixed = $this->getQueue($queue));

        list($job, $reserved) = $this->retrieveNextJob($prefixed);

        if ($reserved) {
            return new RedisJob(
                $this->container, $this, $job,
                $reserved, $this->connectionName, $queue ?: $this->default
            );
        }
    }
}

protected function getQueue($queue)
{
    return 'queues:'.($queue ?: $this->default);
}

在从队列中取出任务之前，需要先将 delay 队列和 reserved 队列中已经到时间的任务放到主队列中：

protected function migrate($queue)
{
    $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()
    );
}

由于从队列取出任务、在队列删除任务、压入主队列是三个操作，为了防止并发，程序这里使用了 LUA 脚本，保证三个操作的原子性：

public static function migrateExpiredJobs()
{
    return <<<'LUA'
          -- Get all of the jobs with an expired "score"...
        local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])

        -- If we have values in the array, we will remove them from the first queue
        -- and add them onto the destination queue in chunks of 100, which moves
        -- all of the appropriate jobs onto the destination queue very safely.
        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;
}

接下来，就要从主队列中获取下一个任务，在取出下一个任务之后，还要将任务放入 reserved 队列中，当任务执行失败后，该任务会进行重试。

protected function retrieveNextJob($queue)
{
    return $this->getConnection()->eval(
        LuaScripts::pop(), 2, $queue, $queue.':reserved',
        $this->availableAt($this->retryAfter)
    );
}

public static function pop()
{
    return <<<'LUA'
        -- Pop the first job off of the queue...
        local job = redis.call('lpop', KEYS[1])
        local reserved = false

        if(job ~= false) then
            -- Increment the attempt count and place job on the reserved queue...
            reserved = cjson.decode(job)
            reserved['attempts'] = reserved['attempts'] + 1
            reserved = cjson.encode(reserved)
            redis.call('zadd', KEYS[2], ARGV[1], reserved)
        end

        return {job, reserved}
    LUA;
}

从 redis 中获取到 job 之后，就会将其包装成 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();
}

public function payload()
{
    return json_decode($this->getRawBody(), true);
}

public function getRawBody()
{
    return $this->job;
}

超时处理

如果一个脚本超时， pcntl_alarm 将会启动并杀死当前的 work 进程。杀死进程后， work 进程将会被守护进程重启，继续进行下一个任务。

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

        pcntl_alarm($this->timeoutForJob($job, $options) + $options->sleep);
    }
}

protected function timeoutForJob($job, WorkerOptions $options)
{
    return $job && ! is_null($job->timeout()) ? $job->timeout() : $options->timeout;
}

任务事务

运行任务前后会启动两个事件 JobProcessing 与 JobProcessed，这两个事件需要事先注册监听者

protected function runJob($job, $connectionName, WorkerOptions $options)
{
    try {
        return $this->process($connectionName, $job, $options);
    } catch (Exception $e) {
        $this->exceptions->report($e);
    } catch (Throwable $e) {
        $this->exceptions->report(new FatalThrowableError($e));
    }
}

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)
        );
    }
}
任务前与任务后事件

raiseBeforeJobEvent 函数用于触发任务处理前的事件，raiseAfterJobEvent 函数用于触发任务处理后的事件：

protected function raiseBeforeJobEvent($connectionName, $job)
{
    $this->events->fire(new Events\JobProcessing(
        $connectionName, $job
    ));
}

protected function raiseAfterJobEvent($connectionName, $job)
{
    $this->events->fire(new Events\JobProcessed(
        $connectionName, $job
    ));
}

任务异常处理

任务在运行过程中会遇到异常情况，这个时候就要判断当前任务的失败次数是不是超过限制。如果没有超过限制，那么就会把当前任务重新放回队列当中；如果超过了限制，那么就要标记当前任务为失败任务，并且将任务从 reserved 队列中删除。

任务失败

markJobAsFailedIfAlreadyExceedsMaxAttempts 函数用于任务运行前，判断当前任务是否重试次数超过限制:

protected function markJobAsFailedIfAlreadyExceedsMaxAttempts($connectionName, $job, $maxTries)
{
    $maxTries = ! is_null($job->maxTries()) ? $job->maxTries() : $maxTries;

    if ($maxTries === 0 || $job->attempts() <= $maxTries) {
        return;
    }

    $this->failJob($connectionName, $job, $e = new MaxAttemptsExceededException(
        'A queued job has been attempted too many times. The job may have previously timed out.'
    ));

    throw $e;
}

public function maxTries()
{
    return array_get($this->payload(), 'maxTries');
}

public function attempts()
{
    return Arr::get($this->decoded, 'attempts') + 1;
}

protected function failJob($connectionName, $job, $e)
{
    return FailingJob::handle($connectionName, $job, $e);
}

当遇到重试次数大于限制的任务，work 进程就会调用 FailingJob:

protected function failJob($connectionName, $job, $e)
{
    return FailingJob::handle($connectionName, $job, $e);
}

public static function handle($connectionName, $job, $e = null)
{
    $job->markAsFailed();

    if ($job->isDeleted()) {
        return;
    }

    try {
        $job->delete();

        $job->failed($e);
    } finally {
        static::events()->fire(new JobFailed(
            $connectionName, $job, $e ?: new ManuallyFailedException
        ));
    }
}

public function markAsFailed()
{
    $this->failed = true;
}

public function delete()
{
    parent::delete();

    $this->redis->deleteReserved($this->queue, $this);
}

public function isDeleted()
{
    return $this->deleted;
}

FailingJob 会标记当前任务 failed、deleted，并且会将当前任务移除 reserved 队列，不会再重试：

public function deleteReserved($queue, $job)
{
    $this->getConnection()->zrem($this->getQueue($queue).':reserved', $job->getReservedJob());
}

FailingJob 还会调用 RedisJob 的 failed 函数，并且触发 JobFailed 事件：

public function failed($e)
{
    $this->markAsFailed();

    $payload = $this->payload();

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

    if (method_exists($this->instance = $this->resolve($class), 'failed')) {
        $this->instance->failed($payload['data'], $e);
    }
}

程序会解析 job 类，我们先前在 redis 中已经存储了：

[
    'job' => 'Illuminate\Queue\CallQueuedHandler@call',
    'maxTries' => isset($job->tries) ? $job->tries : null,
    'timeout' => isset($job->timeout) ? $job->timeout : null,
    'data' => [
        'commandName' => get_class($job),
        'command' => serialize(clone $job),
    ],
];

我们接着看 failed 函数：

public function failed(array $data, $e)
{
    $command = unserialize($data['command']);

    if (method_exists($command, 'failed')) {
        $command->failed($e);
    }
}

可以看到，最后程序调用了任务类的 failed 函数。

异常处理

当任务遇到异常的时候，程序仍然会判断当前任务的重试次数，如果本次任务的重试次数已经大于或等于限制，那么就会停止重试，标记为失败；否则就会重新放入队列，记录日志。

protected function handleJobException($connectionName, $job, WorkerOptions $options, $e)
{
    try {
        $this->markJobAsFailedIfWillExceedMaxAttempts(
            $connectionName, $job, (int) $options->maxTries, $e
        );

        $this->raiseExceptionOccurredJobEvent(
            $connectionName, $job, $e
        );
    } finally {
        if (! $job->isDeleted()) {
            $job->release($options->delay);
        }
    }

    throw $e;
}

protected function markJobAsFailedIfWillExceedMaxAttempts($connectionName, $job, $maxTries, $e)
{
    $maxTries = ! is_null($job->maxTries()) ? $job->maxTries() : $maxTries;

    if ($maxTries > 0 && $job->attempts() >= $maxTries) {
        $this->failJob($connectionName, $job, $e);
    }
}

public function release($delay = 0)
{
    parent::release($delay);

    $this->redis->deleteAndRelease($this->queue, $this, $delay);
}

public function deleteAndRelease($queue, $job, $delay)
{
    $queue = $this->getQueue($queue);

    $this->getConnection()->eval(
        LuaScripts::release(), 2, $queue.':delayed', $queue.':reserved',
        $job->getReservedJob(), $this->availableAt($delay)
    );
}

一旦任务出现异常错误。那么该任务将会立刻从 reserved 队列放入 delayed 队列，并且抛出异常，抛出异常后，程序会将其记录在日志中。

public static function release()
{
    return <<<'LUA'
        -- Remove the job from the current queue...
        redis.call('zrem', KEYS[2], ARGV[1])

        -- Add the job onto the "delayed" queue...
        redis.call('zadd', KEYS[1], ARGV[2], ARGV[1])

        return true
    LUA;
}

任务的运行

任务的运行首先会调用 CallQueuedHandler 的 call 函数：

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

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

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

public function call(Job $job, array $data)
{
    $command = $this->setJobInstanceIfNecessary(
        $job, unserialize($data['command'])
    );

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

    if (! $job->isDeletedOrReleased()) {
        $job->delete();
    }
}

setJobInstanceIfNecessary 函数用于为任务类的 trait: InteractsWithQueue 的设置任务类：

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

    return $instance;
}

trait InteractsWithQueue
{
    public function setJob(JobContract $job)
    {
        $this->job = $job;

        return $this;
    }
}

接着任务的运行就要交给 dispatch :

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

public function getCommandHandler($command)
{
    if ($this->hasCommandHandler($command)) {
        return $this->container->make($this->handlers[get_class($command)]);
    }

    return false;
}

public function hasCommandHandler($command)
{
    return array_key_exists(get_class($command), $this->handlers);
}

如果不对 dispatcher 类进行任何 map 函数设置，getCommandHandler 将会返回 null，此时就会调用任务类的 handle 函数，进行具体的业务逻辑。

任务结束后，就会调用 delete 函数：

public function delete()
{
    parent::delete();

    $this->redis->deleteReserved($this->queue, $this);
}

public function deleteReserved($queue, $job)
{
    $this->getConnection()->zrem($this->getQueue($queue).':reserved', $job->getReservedJob());
}

这样，运行成功的任务会从 reserved 中删除。

————————————————
原文作者：leoyang
转自链接：https://learnku.com/articles/7038/laravel-queue-source-analysis-of-message-queue-task-processor
版权声明：著作权归作者所有。商业转载请联系作者获得授权，非商业转载请保留以上作者信息和原文链接。

 类似资料：

相关阅读

深入理解Java之HashMap源码剖析Sidekiq不处理队列互操作性Azure Service Bus消息队列消息PHP+RabbitMQ实现消息队列的完整代码PHP消息队列实现及应用详解【队列处理订单系统和配送系统】

相关文章

消息中间件 RabbitMQ 之 工作队列（2）—消息应答消息队列相关的5个问题Redis和RabbitMQ做消息队列的区别消息队列 RabbitMQ的介绍及安装面试-消息队列中，如何保证消息的顺序性

相关问答

Redis队列在Rails中处理异步任务源/处理器/接收器消息传递（java.util.HashMap）当队列有n个待处理消息时，ActiveMQ不返回消息RabbitMQ队列消息使用任务执行器的事务轮询器是否将数据库资源用于队列消息

相关文档

libevent 源码深度剖析Mininet 应用与源码剖析Windows MSMQ 消息队列［初步文档］小米消息队列 EMQ 开发文档小米消息队列 Talos 开发文档
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.