发布于 由admin

深入Laravel Queue系统：任务分发、失败重试策略与Horizon监控的底层实现

深入Laravel Queue系统：任务分发、失败重试策略与Horizon监控的底层实现

大家好，今天我们深入探讨Laravel的Queue系统，一个强大且灵活的异步任务处理机制。我们将从任务的分发开始，逐步分析失败重试策略的实现，最后深入了解Horizon监控平台的底层原理。

1. 任务分发：dispatch()方法背后的故事

在Laravel中，我们通常使用dispatch()方法将任务推送到队列。但dispatch()方法背后发生了什么呢？ 它如何将一个简单的类变成一个能在后台执行的任务？

dispatch()方法实际上是一个门面（Facade）调用，最终会调用到IlluminateFoundationBusDispatchable trait中的dispatch()方法。 这个trait被许多Job类使用，提供了便捷的任务分发功能。

<?php

namespace AppJobs;

use IlluminateBusQueueable;
use IlluminateContractsQueueShouldQueue;
use IlluminateFoundationBusDispatchable;
use IlluminateQueueInteractsWithQueue;
use IlluminateQueueSerializesModels;

class ProcessPodcast implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $podcast;

    public function __construct($podcast)
    {
        $this->podcast = $podcast;
    }

    public function handle()
    {
        // 执行处理Podcast的逻辑
        // 例如，生成缩略图，转码等
        Log::info('Processing podcast: ' . $this->podcast);
    }
}

// 使用 dispatch() 方法分发任务
ProcessPodcast::dispatch('My Awesome Podcast');

当我们调用ProcessPodcast::dispatch('My Awesome Podcast')时，发生了以下几个关键步骤：

  • 实例化Job类： 创建ProcessPodcast类的一个实例，传入构造函数所需的参数。
  • 序列化Job实例： SerializesModels trait负责序列化Job实例。 如果Job类的属性是Eloquent模型，它会将模型的主键和类名序列化，而不是整个模型对象。 这避免了在队列中存储大量数据，并确保任务执行时能够重新加载模型。
  • 推送到队列： 将序列化后的Job数据推送到指定的队列连接和队列名称。 Laravel使用配置项queue.php来确定默认的队列连接。 dispatch()方法最终会调用队列驱动（例如redis、database、beanstalkd等）的push()方法。

让我们深入了解SerializesModels trait的序列化过程。

<?php

namespace IlluminateQueue;

use IlluminateContractsQueueQueueableEntity;
use IlluminateDatabaseEloquentModel;
use ReflectionClass;
use ReflectionProperty;

trait SerializesModels
{
    /**
     * Prepare the instance for serialization.
     *
     * @return array
     */
    public function __sleep()
    {
        $properties = (new ReflectionClass($this))->getProperties(ReflectionProperty::IS_PUBLIC | ReflectionProperty::IS_PROTECTED);

        foreach ($properties as $property) {
            $property->setAccessible(true);

            $value = $property->getValue($this);

            if ($value instanceof Model && ! $value instanceof QueueableEntity) {
                $property->setValue($this, $this->getSerializedPropertyValue($value));
            }
        }

        return array_keys(get_object_vars($this));
    }

    /**
     * Restore the model after serialization.
     *
     * @return void
     */
    public function __wakeup()
    {
        $properties = (new ReflectionClass($this))->getProperties(ReflectionProperty::IS_PUBLIC | ReflectionProperty::IS_PROTECTED);

        foreach ($properties as $property) {
            $property->setAccessible(true);

            $value = $property->getValue($this);

            if (is_array($value) && isset($value['class']) && isset($value['id']) && class_exists($value['class'])) {
                $property->setValue($this, $this->getRestoredPropertyValue($value));
            }
        }
    }

    /**
     * Get the serialized representation of a property.
     *
     * @param  mixed  $value
     * @return array
     */
    protected function getSerializedPropertyValue($value)
    {
        return [
            'class' => get_class($value),
            'id' => $value->getKey(),
            'relations' => $value->relations,
            'connection' => $value->getConnectionName(),
        ];
    }

    /**
     * Get the restored property value after serialization.
     *
     * @param  array  $value
     * @return mixed
     */
    protected function getRestoredPropertyValue(array $value)
    {
        $class = $value['class'];
        $id = $value['id'];
        $connection = $value['connection'];

        $model = (new $class)->setConnection($connection)->find($id);

        if (isset($value['relations']) && is_array($value['relations'])) {
            $model->setRelations($value['relations']);
        }

        return $model;
    }
}

__sleep()方法在序列化之前被调用，它会遍历Job类的所有属性，如果属性是Eloquent模型，则将其替换为一个包含模型类名、主键ID、关联关系和连接信息的数组。 __wakeup()方法在反序列化之后被调用，它会根据序列化的信息重新加载模型。

2. 失败重试策略：tries, timeout, 和 retryAfter

任务执行过程中可能会因为各种原因失败，例如网络问题、数据库连接错误等。 Laravel提供了多种机制来处理失败的任务，确保任务最终能够完成。

  • tries属性： 定义了任务可以被尝试的最大次数。 可以在Job类中定义$tries属性来设置重试次数。 如果任务失败次数超过$tries，则会被标记为失败。
  • timeout属性： 定义了任务执行的最大时长（秒）。 如果任务执行时间超过timeout，则会被认为失败。 同样可以在Job类中定义$timeout属性。
  • retryAfter属性： 定义了在任务失败后，等待多久（秒）再进行重试。 这个属性主要用于限流场景，例如在调用外部API时，如果API返回错误，可以设置retryAfter来避免过快地重试。
<?php

namespace AppJobs;

use IlluminateBusQueueable;
use IlluminateContractsQueueShouldQueue;
use IlluminateFoundationBusDispatchable;
use IlluminateQueueInteractsWithQueue;
use IlluminateQueueSerializesModels;
use Throwable;

class ProcessPodcast implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $podcast;

    public $tries = 3; // 最多重试3次
    public $timeout = 60; // 任务超时时间为60秒

    public function __construct($podcast)
    {
        $this->podcast = $podcast;
    }

    public function handle()
    {
        // 模拟任务失败
        if (rand(0, 1) == 0) {
            throw new Exception('Podcast processing failed!');
        }

        Log::info('Processing podcast: ' . $this->podcast);
    }

    /**
     * 处理一个失败的任务。
     *
     * @param  Throwable  $exception
     * @return void
     */
    public function failed(Throwable $exception)
    {
        // 发送失败通知
        Log::error('Podcast processing failed: ' . $this->podcast . ' - ' . $exception->getMessage());
    }
}

当任务失败时，Laravel会将其重新放回队列。 队列驱动会根据retryAfter属性来决定何时重新执行任务。 如果任务失败次数超过tries，则会调用Job类的failed()方法（如果定义了）。 failed()方法可以用于执行一些清理工作，例如发送失败通知、记录错误日志等。

失败任务的处理机制

Laravel提供了两种主要的方式来处理失败的任务：

  • 失败任务表： Laravel可以将失败的任务记录到failed_jobs表中。 可以使用php artisan queue:failed-table命令创建迁移文件。 然后使用php artisan migrate命令执行迁移。 一旦任务达到最大重试次数，它将被插入到failed_jobs表中。 可以使用php artisan queue:retry all 或者 php artisan queue:retry <id>来重试失败的任务。
  • 自定义失败处理逻辑： 可以在Job类中定义failed()方法来执行自定义的失败处理逻辑。 如上面的代码所示，failed()方法接收一个Throwable对象，可以访问任务失败的原因。

队列驱动的重试机制

不同的队列驱动实现重试机制的方式可能略有不同。 例如：

  • Redis: Redis队列通常使用延迟队列来实现重试。 当任务失败时，将其放入一个延迟队列，等待一段时间后再将其放回主队列。
  • Database: Database队列通常通过更新数据库记录来实现重试。 当任务失败时，更新任务记录的attempts字段，并设置available_at字段为重试时间。

3. Horizon监控：底层实现与核心组件

Laravel Horizon 是一个基于 Redis 的队列监控和管理平台。 它提供了一个漂亮的仪表盘，可以实时查看队列状态、任务执行情况、失败任务等信息。 Horizon的核心在于监控、分析和管理Laravel队列系统。

核心组件

  • Supervisor： Horizon 使用 Supervisor 作为进程管理器。 Supervisor 负责启动、停止和监控 Horizon 的 worker 进程。
  • Redis: Horizon 使用 Redis 存储队列状态、任务执行信息、指标数据等。
  • Horizon CLI： Horizon 提供了一组命令行工具，用于管理 Horizon 进程、查看队列状态等。
  • Horizon UI： Horizon 提供了一个基于 Laravel 的 Web UI，用于实时监控队列状态、任务执行情况、失败任务等信息。

Horizon 的工作流程

  1. 启动 Horizon： 使用 php artisan horizon 命令启动 Horizon。 该命令会启动 Supervisor，Supervisor 会根据配置文件启动 Horizon 的 worker 进程。
  2. Worker 进程： Horizon 的 worker 进程负责从 Redis 队列中拉取任务，并执行任务。
  3. 监控数据： Worker 进程会将任务执行信息、队列状态等数据存储到 Redis 中。
  4. UI 显示： Horizon 的 Web UI 从 Redis 中读取数据，并将其展示在仪表盘上。

Horizon 的核心功能

  • 队列监控： 实时监控队列的长度、吞吐量、平均处理时间等指标。
  • 任务监控： 监控任务的执行状态、执行时间、失败次数等信息。
  • 失败任务管理： 查看失败任务的详细信息，并可以重试失败的任务。
  • 进程管理： 管理 Horizon 的 worker 进程，例如启动、停止、重启进程。
  • 指标分析： 分析队列和任务的性能指标，找出瓶颈，并进行优化。
  • 标签功能： 可以为任务添加标签，方便分组和筛选。
  • 沉默期间 (Quiet Time) 与等待时间 (Wait Time): Horizon 会根据队列的负载动态调整 worker 进程的数量。

Horizon 的底层实现

Horizon 的底层实现主要涉及以下几个方面：

  • 数据存储： Horizon 使用 Redis 的 List、Hash、Set 等数据结构来存储队列状态、任务执行信息、指标数据等。 例如，使用 List 存储队列中的任务，使用 Hash 存储任务的详细信息，使用 Set 存储正在执行的任务的 ID。
  • 数据采集： Horizon 的 worker 进程在执行任务时，会采集任务的执行时间、状态等信息，并将这些信息存储到 Redis 中。
  • 数据分析： Horizon 会定期从 Redis 中读取数据，并进行分析，生成队列的长度、吞吐量、平均处理时间等指标。
  • UI 展示： Horizon 的 Web UI 使用 Vue.js 和 Laravel Echo 来实时更新仪表盘上的数据。

Horizon 的配置

Horizon 的配置文件位于 config/horizon.php。 可以在该文件中配置 Horizon 的连接信息、队列设置、进程数量等。

<?php

use LaravelHorizonContractsHorizonCommand;

return [

    /*
    |--------------------------------------------------------------------------
    | Horizon Redis Connection
    |--------------------------------------------------------------------------
    |
    | This is the connection in your "database" configuration that should be
    | used by Laravel Horizon to pull job information and monitor your
    | queues. When using Redis, this should be the name of your redis
    | connection.
    |
    */

    'use' => 'redis',

    /*
    |--------------------------------------------------------------------------
    | Horizon Redis Prefix
    |--------------------------------------------------------------------------
    |
    | This prefix will be used when storing all Horizon data in Redis. You
    | may modify the prefix when you are running multiple Horizon instances
    | on the same Redis server so that they don't have overlapping data.
    |
    */

    'prefix' => env('HORIZON_PREFIX', 'horizon:'),

    /*
    |--------------------------------------------------------------------------
    | Horizon Route Middleware
    |--------------------------------------------------------------------------
    |
    | These middleware will be assigned to the Horizon route, giving you
    | the ability to define some extra authorization to your Horizon UI.
    | When using the default "web" middleware, be sure to also include
    | the "AuthenticateSession" middleware to maintain your auth state.
    |
    */

    'middleware' => ['web', 'auth'],

    /*
    |--------------------------------------------------------------------------
    | Horizon Path
    |--------------------------------------------------------------------------
    |
    | This is the URI path where Horizon will be accessible from. Feel free
    | to change this path to anything you like. Note that the URI will not
    | affect the paths of your actual API endpoints, as that is up to you.
    |
    */

    'path' => 'horizon',

    /*
    |--------------------------------------------------------------------------
    | Horizon Dashboard Configuration
    |--------------------------------------------------------------------------
    |
    | Here you may configure some display settings for Horizon.
    |
    */

    'sort' => 'waits',

    'queues' => [
        'default' => [
            'connection' => 'redis',
            'queue' => 'default',
            'balance' => 'auto',
            'maxProcesses' => 10,
            'tries' => 3,
            'timeout' => 60,
        ],

        'emails' => [
            'connection' => 'redis',
            'queue' => 'emails',
            'balance' => 'simple',
            'maxProcesses' => 5,
            'tries' => 3,
            'timeout' => 60,
        ],
    ],

    /*
    |--------------------------------------------------------------------------
    | Horizon Notification Channels
    |--------------------------------------------------------------------------
    |
    | Here you may configure the notification channels that will be used to
    | notify you of important Horizon events, such as long wait times or
    | job failures.
    |
    */

    'notifications' => [
        'waits' => ['slack'],
        'failures' => ['slack'],
    ],

    /*
    |--------------------------------------------------------------------------
    | Horizon Wait Time Thresholds
    |--------------------------------------------------------------------------
    |
    | This option allows you to configure when Horizon begin indicating that
    | the wait times on your queues are too high. You may define a single
    | value for all queues or specify a specific value for each queue.
    |
    */

    'waits' => [
        'default' => 60,
        'redis:emails' => 120,
    ],

    /*
    |--------------------------------------------------------------------------
    | Horizon Trim Settings
    |--------------------------------------------------------------------------
    |
    | Here you can configure how many recent jobs Horizon should keep tabs
    | on before removing them from the view. Each option should be given
    | in minutes.
    |
    */

    'trim' => [
        'recent' => 60,
        'pending' => 60,
        'completed' => 60,
        'failed' => 10080,
        'monitored' => 10080,
    ],
];

自定义 Horizon 指标

可以自定义 Horizon 指标，例如监控特定任务的执行时间、成功率等。 需要实现 LaravelHorizonContractsHorizonCommand 接口，并将其注册到 Horizon 中。

<?php

namespace AppConsoleCommands;

use IlluminateConsoleCommand;
use LaravelHorizonContractsHorizonCommand;

class MonitorCustomTask implements HorizonCommand
{
    protected $signature = 'horizon:monitor-task {task}';

    protected $description = 'Monitor the execution time of a custom task';

    public function handle()
    {
        $task = $this->argument('task');

        // 监控任务执行时间
        $startTime = microtime(true);

        // 执行任务逻辑

        $endTime = microtime(true);

        $executionTime = $endTime - $startTime;

        // 将执行时间存储到 Redis 中

        // 可以将任务状态、成功率等信息也存储到 Redis 中
    }
}

然后需要在 config/app.php 文件的 providers 数组中注册该命令。

4. 总结：任务队列的艺术

我们深入了解了Laravel Queue系统的核心机制，从任务的分发、序列化，到失败重试策略和Horizon监控的底层实现。掌握这些知识，能更好地理解和利用Laravel Queue，构建高效、可靠的异步任务处理系统。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注