队列

简介

在构建 Web 应用的过程中，你可能会遇到一些任务，例如解析和存储上传的 CSV 文件，在常规的 Web 请求中执行这些任务耗时过长。好在，Laravel 允许你轻松创建可在后台处理的排队任务。通过将耗时任务转移到队列中，你的应用能够以极快的速度响应 Web 请求，并为你的客户带来更好的用户体验。

Laravel 队列针对多种不同的队列后端，例如 Amazon SQS、Redis，甚至关系型数据库，提供了统一的队列 API。

Laravel 的队列配置选项存储在你应用的 config/queue.php 配置文件中。在该文件中，你会找到框架内置的每个队列驱动的连接配置，包括数据库、Amazon SQS、Redis 和 Beanstalkd 驱动，还有一个会立即执行任务的同步驱动（用于开发或测试场景）。此外，还包含一个 null 队列驱动，它会丢弃排队的任务。

Laravel Horizon 是一个为基于 Redis 的队列提供的美观的仪表盘和配置系统。如需更多信息，请查看完整的 Horizon 文档。

连接与队列

在使用 Laravel 队列之前，理解“连接”与“队列”的区别至关重要。在你的 config/queue.php 配置文件中，有一个 connections 配置数组。该选项定义了与后端队列服务（如 Amazon SQS、Beanstalk 或 Redis）的连接。不过，任意一个队列连接都可以包含多个“队列”，这些队列可以被看作是排队任务的不同堆栈或堆叠。

请注意，queue 配置文件中的每个连接配置示例都包含一个 queue 属性。这是任务被发送到指定连接时，默认会被分发到的队列。换句话说，如果你分发任务时没有明确指定它应被分发到哪个队列，该任务会被放入连接配置的 queue 属性所定义的队列中：

use App\Jobs\ProcessPodcast;

// This job is sent to the default connection's default queue...
ProcessPodcast::dispatch();

// This job is sent to the default connection's "emails" queue...
ProcessPodcast::dispatch()->onQueue('emails');

有些应用可能不需要将任务推送到多个队列，而是更倾向于使用一个简单的队列。不过，将任务推送到多个队列，对于希望对任务的处理方式进行优先级划分或分类的应用而言特别有用，因为 Laravel 队列工作进程允许你指定它应按优先级处理哪些队列。例如，如果你将任务推送到 high 队列，你可以运行一个赋予这些任务更高处理优先级的工作进程：

php artisan queue:work --queue=high,default

驱动说明与前提条件

数据库

若要使用 database 队列驱动，你需要一个数据库表来存放任务。通常，该表包含在 Laravel 默认的 0001_01_01_000002_create_jobs_table.php 数据库迁移文件 中；不过，如果你的应用不包含此迁移文件，你可以使用 make:queue-table Artisan 命令来创建它：

php artisan make:queue-table

php artisan migrate

Redis

若要使用 redis 队列驱动，你需要在 config/database.php 配置文件中配置 Redis 数据库连接。

redis 队列驱动不支持 serializer 和 compression Redis 选项。

Redis 集群

如果你的 Redis 队列连接使用的是 Redis Cluster，你的队列名称必须包含 键哈希标签。这是必需的，以确保给定队列的所有 Redis 键都被放入同一个哈希槽中：

'redis' => [
    'driver' => 'redis',
    'connection' => env('REDIS_QUEUE_CONNECTION', 'default'),
    'queue' => env('REDIS_QUEUE', '{default}'),
    'retry_after' => env('REDIS_QUEUE_RETRY_AFTER', 90),
    'block_for' => null,
    'after_commit' => false,
],

阻塞

使用 Redis 队列时，你可以使用 block_for 配置选项来指定驱动在进入工作进程循环并重新轮询 Redis 数据库之前，应该等待多久以获取可用任务。

根据队列负载调整该值，比持续轮询 Redis 数据库来获取新任务更为高效。例如，你可以将该值设置为 5，表示驱动在等待任务可用时应阻塞五秒钟：

'redis' => [
    'driver' => 'redis',
    'connection' => env('REDIS_QUEUE_CONNECTION', 'default'),
    'queue' => env('REDIS_QUEUE', 'default'),
    'retry_after' => env('REDIS_QUEUE_RETRY_AFTER', 90),
    'block_for' => 5,
    'after_commit' => false,
],

将 block_for 设置为 0，会导致队列工作进程无限阻塞，直到有任务可用。这也会导致像 SIGTERM 这类信号无法得到处理，直到下一个任务被处理完毕。

SQS 溢出存储

Amazon SQS 限制了排队消息负载的最大大小。如果你需要分发的负载大小可能超过该限制的任务，你可以配置 Laravel，将超大的 SQS 负载存储在缓存存储中，转而通过 SQS 发送指针。要启用此功能，请在 SQS 队列连接配置中添加一个 overflow 数组：

'sqs' => [
    'driver' => 'sqs',
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'prefix' => env('SQS_PREFIX', 'https://sqs.us-east-1.amazonaws.com/your-account-id'),
    'queue' => env('SQS_QUEUE', 'default'),
    'suffix' => env('SQS_SUFFIX'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'after_commit' => false,
    'overflow' => [
        'enabled' => env('SQS_OVERFLOW_ENABLED', false),
        'store' => env('SQS_OVERFLOW_STORE'),
        'always' => false,
        'delete_after_processing' => true,
        'flush_on_clear' => env('SQS_OVERFLOW_FLUSH_ON_CLEAR', false),
    ],
],

启用溢出存储后，Laravel 会将大小至少为 1 MB 的负载存储到已配置的缓存存储中。如果 always 选项为 true，无论大小如何，每个 SQS 负载都会被存储到缓存存储中。由于排队任务在被处理时需要从缓存存储中检索其负载，你应该选择一个能够在工作进程处理负载之前一直保留负载的存储方式。默认情况下，存储的负载在对应的任务成功处理并从 SQS 中删除后，也会被删除。

如果 flush_on_clear 选项为 true，当 queue:clear 命令清空 SQS 队列时，配置的溢出缓存存储会被清空。由于清空缓存存储可能会移除该存储中的所有项，因此在启用此选项时，应将 SQS 溢出存储配置为使用专用的缓存存储。

其他驱动的前提条件

以下队列驱动需要相应的依赖。这些依赖可以通过 Composer 包管理器安装：

• Amazon SQS：aws/aws-sdk-php ~3.0
• Beanstalkd：pda/pheanstalk ^7.0|^8.0
• Redis：predis/predis ~3.0 或 phpredis PHP 扩展
• MongoDB：mongodb/laravel-mongodb

创建任务

生成任务类

默认情况下，你应用中所有可排队的任务都存储在 app/Jobs 目录中。如果 app/Jobs 目录不存在，在运行 make:job Artisan 命令时会创建该目录：

php artisan make:job ProcessPodcast

生成的类会实现 Illuminate\Contracts\Queue\ShouldQueue 接口，以此告知 Laravel 该任务应被推送到队列中以异步运行。

可以通过 桩文件发布来自定义任务桩文件。

类结构

任务类非常简单，通常只包含一个 handle 方法，该方法会在任务被队列处理时被调用。首先，让我们看一个任务类的示例。在此示例中，我们假设自己运营着一个播客发布服务，需要在发布上传的播客文件之前对这些文件进行处理：

<?php

namespace App\Jobs;

use App\Models\Podcast;
use App\Services\AudioProcessor;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * 创建新的作业实例。
     */
    public function __construct(
        public Podcast $podcast,
    ) {}

    /**
     * 执行作业。
     */
    public function handle(AudioProcessor $processor): void
    {
        // 处理已上传的播客...
    }
}

在此示例中，请注意我们可以直接将Eloquent 模型传入队列作业的构造函数。由于作业使用了 Queueable 特性，在作业处理时，Eloquent 模型及其已加载的关系将被妥善地序列化和反序列化。

如果你的队列作业在其构造函数中接收一个 Eloquent 模型，那么只有该模型的标识符会被序列化到队列中。当作业实际被处理时，队列系统会自动从数据库中重新检索完整的模型实例及其已加载的关系。这种模型序列化方式可以向队列驱动发送更小的作业负载。

handle 方法依赖注入

当作业被队列处理时，会调用 handle 方法。请注意，我们可以在作业的 handle 方法上对依赖项进行类型提示。Laravel 服务容器会自动注入这些依赖项。

如果你想完全控制容器如何将依赖项注入到 handle 方法中，可以使用容器的 bindMethod 方法。bindMethod 方法接收一个回调函数，该回调函数会接收作业和容器。在回调函数中，你可以随意调用 handle 方法。通常，你应该在 App\Providers\AppServiceProvider 服务提供者的 boot 方法中调用此方法：

use App\Jobs\ProcessPodcast;
use App\Services\AudioProcessor;
use Illuminate\Contracts\Foundation\Application;

$this->app->bindMethod([ProcessPodcast::class, 'handle'], function (ProcessPodcast $job, Application $app) {
    return $job->handle($app->make(AudioProcessor::class));
});

二进制数据，如原始图像内容，在传递给队列作业之前，应先通过 base64_encode 函数处理。否则，该作业在放入队列时可能无法正确序列化为 JSON。

队列关系

由于所有已加载的 Eloquent 模型关系在作业被加入队列时也会被序列化，序列化后的作业字符串有时会变得非常大。此外，当作业被反序列化并从数据库中重新检索模型关系时，会检索完整的关系。在作业排队过程中模型被序列化之前应用的任何关系约束，在作业被反序列化时都不会被应用。因此，如果你想处理特定关系的一个子集，应该在队列作业中重新对该关系进行约束。

或者，为了避免关系被序列化，在设置属性值时，可以对模型调用 withoutRelations 方法。该方法会返回一个不包含已加载关系的模型实例：

/**
 * 创建新的作业实例。
 */
public function __construct(
    Podcast $podcast,
) {
    $this->podcast = $podcast->withoutRelations();
}

如果你只想移除特定的关系，同时保留其他关系，可以使用 withoutRelation 方法：

$this->podcast = $podcast->withoutRelation('comments');

如果你使用PHP 构造函数属性提升，并且想表明某个 Eloquent 模型不应序列化其关系，可以使用 WithoutRelations 特性：

use Illuminate\Queue\Attributes\WithoutRelations;

/**
 * 创建新的作业实例。
 */
public function __construct(
    #[WithoutRelations]
    public Podcast $podcast,
) {}

为了方便起见，如果你想序列化所有不包含关系的模型，可以将 WithoutRelations 特性应用于整个类，而不必对每个模型单独应用该特性：

<?php

namespace App\Jobs;

use App\Models\DistributionPlatform;
use App\Models\Podcast;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\WithoutRelations;

#[WithoutRelations]
class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * 创建新的作业实例。
     */
    public function __construct(
        public Podcast $podcast,
        public DistributionPlatform $platform,
    ) {}
}

如果作业接收的是一个 Eloquent 模型的集合或数组，而非单个模型，那么在作业被反序列化并执行时，该集合中的模型不会恢复其关系。这是为了防止处理大量模型的作业过度消耗资源。

唯一作业

唯一作业需要缓存驱动支持锁。目前，memcached、redis、dynamodb、database、file 和 array 缓存驱动支持原子锁。

唯一作业约束不适用于批次内的作业。

有时，你可能希望确保在任何时刻，队列中只有一个特定作业的实例。你可以通过在作业类中实现 ShouldBeUnique 接口来实现这一点。该接口不要求你在类中定义任何额外的方法：

<?php

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUnique;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    // ...
}

在上面的示例中，UpdateSearchIndex 作业是唯一的。因此，如果队列中已经存在该作业的另一个实例且尚未完成处理，该作业将不会被分发。

在某些情况下，你可能想定义一个特定的“键”来使作业唯一，或者你可能想指定一个超时时间，超过该时间作业就不再保持唯一。为此，你可以使用 UniqueFor 特性，并在作业类中定义一个 uniqueId 方法：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Queue\Attributes\UniqueFor;

#[UniqueFor(3600)]
class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    /**
     * 产品实例。
     *
     * @var \App\Models\Product
     */
    public $product;

    /**
     * 获取作业的唯一 ID。
     */
    public function uniqueId(): string
    {
        return $this->product->id;
    }
}

在上面的示例中，UpdateSearchIndex 作业根据产品 ID 保持唯一。因此，在现有作业完成处理之前，任何使用相同产品 ID 的新作业分发都会被忽略。此外，如果现有作业在一小时内未被处理，唯一锁将被释放，另一个具有相同唯一键的作业可以被分发到队列中。

如果你的应用从多个 Web 服务器或容器分发作业，你应该确保所有服务器都与同一个中央缓存服务器通信，以便 Laravel 能够准确判断作业是否唯一。

在处理开始前保持作业唯一性

默认情况下，唯一作业在完成处理或所有重试尝试失败后会被“解锁”。然而，在某些情况下，你可能希望作业在被处理之前立即解锁。为此，你的作业应实现 ShouldBeUniqueUntilProcessing 契约，而不是 ShouldBeUnique 契约：

<?php

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUniqueUntilProcessing
{
    // ...
}

唯一作业锁

在后台，当 ShouldBeUnique 作业被分发时，Laravel 会尝试使用 uniqueId 键获取锁。如果该锁已被占用，作业将不会被分发。当作业完成处理或所有重试尝试失败时，此锁将被释放。默认情况下，Laravel 会使用默认的缓存驱动来获取此锁。然而，如果你希望使用其他驱动来获取锁，可以定义一个 uniqueVia 方法，该方法返回应使用的缓存驱动：

use Illuminate\Contracts\Cache\Repository;
use Illuminate\Support\Facades\Cache;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    // ...

    /**
     * 获取用于唯一任务锁的缓存驱动程序。
     */
    public function uniqueVia(): Repository
    {
        return Cache::driver('redis');
    }
}

如果你只需要限制任务的并发处理，请改用 WithoutOverlapping 任务中间件。

去抖动任务

有时，你可能需要确保在同一个任务在短时间内多次派发时，仅最新一次派发的任务会被实际执行。你可以通过为任务添加 DebounceFor 特性来实现这一点：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\DebounceFor;

#[DebounceFor(30)]
class UpdateSearchIndex implements ShouldQueue
{
    use Queueable;

    /**
     * 创建新任务实例。
     */
    public function __construct(public int $productId)
    {
    }

    /**
     * 获取任务的去抖动 ID。
     */
    public function debounceId(): string
    {
        return (string) $this->productId;
    }
}

在上面的示例中，在 30 秒内为同一产品重复派发 UpdateSearchIndex 会对任务进行去抖动，从而仅运行最新一次派发的任务。

如果你想要限制频繁重新派发的任务可被延迟的最长时间，可以向 DebounceFor 特性提供 maxWait 参数：

#[DebounceFor(30, maxWait: 120)]
class UpdateSearchIndex implements ShouldQueue
{
    use Queueable;

    // ...
}

你可以通过在任务中定义 debounceVia 方法，来自定义用于去抖动跟踪的缓存存储：

use Illuminate\Contracts\Cache\Repository;
use Illuminate\Support\Facades\Cache;

public function debounceVia(): Repository
{
    return Cache::driver('redis');
}

如果某个去抖动任务被更新的派发所取代，Laravel 将派发 Illuminate\Queue\Events\JobDebounced 事件，并将被取代的任务从队列中移除。

去抖动任务和唯一任务是互斥的。使用 DebounceFor 特性的任务不应实现 ShouldBeUnique。

如果你的应用在多个 Web 服务器或容器中派发去抖动任务，应确保所有服务器都与同一中央缓存服务器通信。

加密任务

Laravel 允许你通过加密来确保任务数据的隐私性和完整性。要开始使用，只需将 ShouldBeEncrypted 接口添加到任务类中。一旦将该接口添加到类中，Laravel 会在将任务推送到队列之前自动对其进行加密：

<?php

use Illuminate\Contracts\Queue\ShouldBeEncrypted;
use Illuminate\Contracts\Queue\ShouldQueue;

class UpdateSearchIndex implements ShouldQueue, ShouldBeEncrypted
{
    // ...
}

任务中间件

任务中间件允许你在队列任务的执行过程中包裹自定义逻辑，从而减少任务本身的样板代码。例如，考虑下面利用 Laravel 的 Redis 速率限制功能的 handle 方法，它仅允许每五秒处理一个任务：

use Illuminate\Support\Facades\Redis;

/**
 * 执行任务。
 */
public function handle(): void
{
    Redis::throttle('key')->block(0)->allow(1)->every(5)->then(function () {
        info('Lock obtained...');

        // Handle job...
    }, function () {
        // Could not obtain lock...

        return $this->release(5);
    });
}

虽然这段代码是有效的，但 handle 方法的实现会变得冗杂，因为它充斥着 Redis 速率限制逻辑。此外，对于我们想要进行速率限制的任何其他任务，这段速率限制逻辑都必须重复编写。与其在 handle 方法中进行速率限制，不如定义一个处理速率限制的任务中间件：

<?php

namespace App\Jobs\Middleware;

use Closure;
use Illuminate\Support\Facades\Redis;

class RateLimited
{
    /**
     * 处理队列任务。
     *
     * @param  \Closure(object): void  $next
     */
    public function handle(object $job, Closure $next): void
    {
        Redis::throttle('key')
            ->block(0)->allow(1)->every(5)
            ->then(function () use ($job, $next) {
                // Lock obtained...

                $next($job);
            }, function () use ($job) {
                // Could not obtain lock...

                $job->release(5);
            });
    }
}

如你所见，与路由中间件类似，任务中间件会接收正在处理的任务以及一个用于继续处理该任务的回调函数。

你可以使用 make:job-middleware Artisan 命令生成新的任务中间件类。创建任务中间件后，可以通过从任务的 middleware 方法中返回它们，将其附加到任务上。make:job Artisan 命令生成的任务脚手架中并不存在该方法，因此你需要手动将其添加到任务类中：

use App\Jobs\Middleware\RateLimited;

/**
 * 获取任务应通过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited];
}

任务中间件还可以分配给可队列化的事件监听器、Mailables以及通知。

速率限制

尽管我们刚刚演示了如何编写自定义的速率限制任务中间件，但实际上 Laravel 自带了一个速率限制中间件，你可以利用它对任务进行速率限制。与路由速率限制器类似，任务速率限制器使用 RateLimiter Facade 的 for 方法来定义。

例如，你可能希望允许普通用户每小时备份一次数据，而对高级客户不设此类限制。为此，你可以在 AppServiceProvider 的 boot 方法中定义一个 RateLimiter：

use Illuminate\Cache\RateLimiting\Limit;
use Illuminate\Support\Facades\RateLimiter;

/**
 * 引导任何应用服务。
 */
public function boot(): void
{
    RateLimiter::for('backups', function (object $job) {
        return $job->user->vipCustomer()
            ? Limit::none()
            : Limit::perHour(1)->by($job->user->id);
    });
}

在上面的示例中，我们定义了一个每小时的速率限制；不过，你可以使用 perMinute 方法轻松定义基于分钟的速率限制。此外，你可以将任何期望的值传递给速率限制的 by 方法；不过，该值通常用于按客户对速率限制进行分段：

return Limit::perMinute(50)->by($job->user->id);

定义好速率限制后，你可以使用 Illuminate\Queue\Middleware\RateLimited 中间件将速率限制器附加到任务上。每次任务超过速率限制时，该中间件会根据速率限制的时长，将任务以适当的延迟重新释放回队列：

use Illuminate\Queue\Middleware\RateLimited;

/**
 * 获取任务应通过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited('backups')];
}

将受速率限制的任务重新释放到队列中仍会增加任务的 attempts 总次数。你可能需要相应地调整任务类上的 Tries 和 MaxExceptions 特性。或者，你可以使用 retryUntil 方法来定义任务不再尝试重试的时间。

使用 releaseAfter 方法，你还可以指定释放的任务在再次尝试之前必须经过的秒数：

/**
 * 获取任务应通过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->releaseAfter(60)];
}

如果你不希望任务在达到速率限制时被重试，可以使用 dontRelease 方法：

/**
 * 获取任务应通过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->dontRelease()];
}

基于 Redis 的速率限制

如果你使用的是 Redis，可以使用 Illuminate\Queue\Middleware\RateLimitedWithRedis 中间件，该中间件针对 Redis 进行了优化，比基本的速率限制中间件更高效：

use Illuminate\Queue\Middleware\RateLimitedWithRedis;

public function middleware(): array
{
    return [new RateLimitedWithRedis('backups')];
}

可以使用 connection 方法指定中间件应使用的 Redis 连接：

return [(new RateLimitedWithRedis('backups'))->connection('limiter')];

防止任务重叠

Laravel 包含 Illuminate\Queue\Middleware\WithoutOverlapping 中间件，可让您基于任意键防止任务重叠。当队列任务正在修改一项一次只能由一个任务修改的资源时，这会很有用。

例如，假设您有一个用于更新用户信用分的队列任务，并且您想要防止针对同一用户 ID 的信用分更新任务重叠。为此，您可以从任务的 middleware 方法返回 WithoutOverlapping 中间件：

use Illuminate\Queue\Middleware\WithoutOverlapping;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new WithoutOverlapping($this->user->id)];
}

将重叠的任务释放回队列时，仍会增加该任务的总尝试次数。您可能需要相应地调整任务类上的 Tries 和 MaxExceptions 属性。例如，如果将 Tries 保留为默认值 1，则任何重叠的任务之后都不会被重试。

同一类型的任何重叠任务都会被释放回队列。您还可以指定释放的任务在再次尝试之前必须经过的秒数：

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->releaseAfter(60)];
}

如果您希望立即删除任何重叠任务，使其不会被重试，可以使用 dontRelease 方法：

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->dontRelease()];
}

WithoutOverlapping 中间件由 Laravel 的原子锁功能提供支持。有时，您的任务可能会意外失败或超时，导致锁未被释放。因此，您可以使用 expireAfter 方法显式定义锁的过期时间。例如，下面的示例将指示 Laravel 在任务开始处理三分钟后释放 WithoutOverlapping 锁：

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->expireAfter(180)];
}

WithoutOverlapping 中间件需要支持锁的缓存驱动。目前，memcached、redis、dynamodb、database、file 和 array 缓存驱动支持原子锁。

跨任务类共享锁键

默认情况下，WithoutOverlapping 中间件仅防止同一类的重叠任务。因此，即使两个不同的任务类可能使用相同的锁键，它们也不会被阻止重叠。不过，您可以使用 shared 方法指示 Laravel 跨任务类应用该键：

use Illuminate\Queue\Middleware\WithoutOverlapping;

class ProviderIsDown
{
    // ...

    public function middleware(): array
    {
        return [
            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
        ];
    }
}

class ProviderIsUp
{
    // ...

    public function middleware(): array
    {
        return [
            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
        ];
    }
}

限制异常

Laravel 包含 Illuminate\Queue\Middleware\ThrottlesExceptions 中间件，可让您限制异常。一旦任务抛出指定数量的异常，所有进一步执行该任务的尝试都会延迟，直到经过指定的时间间隔。此中间件对于与不稳定的第三方服务交互的任务特别有用。

例如，假设有一个队列任务与第三方 API 交互，而该 API 开始抛出异常。要限制异常，您可以从任务的 middleware 方法返回 ThrottlesExceptions 中间件。通常，此中间件应与实现了基于时间的尝试的任务搭配使用：

use DateTime;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new ThrottlesExceptions(10, 5 * 60)];
}

/**
 * 确定任务应该超时的时间。
 */
public function retryUntil(): DateTime
{
    return now()->plus(minutes: 30);
}

该中间件接受的第一个构造函数参数是任务在被限制前可抛出的异常数量，第二个构造函数参数是在任务被限制后，再次尝试该任务之前应经过的秒数。在上面的代码示例中，如果任务连续抛出 10 次异常，我们将等待 5 分钟再尝试该任务，且受限于 30 分钟的时间限制。

当任务抛出异常但尚未达到异常阈值时，任务通常会被立即重试。不过，在将中间件附加到任务时，您可以通过调用 backoff 方法指定此类任务应延迟的分钟数：

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 5 * 60))->backoff(5)];
}

backoff 方法也接受一个接收抛出异常的闭包，允许动态确定延迟：

use App\Exceptions\RateLimitedException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;
use Throwable;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 5 * 60))->backoff(
        fn (Throwable $throwable) => $throwable instanceof RateLimitedException
            ? $throwable->retryAfterMinutes()
            : 5
    )];
}

在内部，此中间件使用 Laravel 的缓存系统来实现速率限制，并将任务的类名用作缓存“键”。在将中间件附加到任务时，您可以通过调用 by 方法覆盖此键。如果您有多个任务与同一个第三方服务交互，并且希望它们共享一个公共的限流“桶”，确保它们遵守单个共享限制，这会很有用：

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->by('key')];
}

默认情况下，此中间件会限制所有异常。您可以通过在将中间件附加到任务时调用 when 方法来修改此行为。只有当提供给 when 方法的闭包返回 true 时，异常才会被限制：

use Illuminate\Http\Client\HttpClientException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->when(
        fn (Throwable $throwable) => $throwable instanceof HttpClientException
    )];
}

与将任务释放回队列或抛出异常的 when 方法不同，deleteWhen 方法允许您在发生特定异常时完全删除该任务：

use App\Exceptions\CustomerDeletedException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * 获取任务应经过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(2, 10 * 60))->deleteWhen(CustomerDeletedException::class)];
}

如果您希望将受限的异常报告给应用程序的异常处理器，可以在将中间件附加到任务时通过调用 report 方法来实现。您可以选择性地为 report 方法提供一个闭包，只有当给定闭包返回 true 时，异常才会被报告：

use Illuminate\Http\Client\HttpClientException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * 获取作业应通过的中间件。
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->report(
        fn (Throwable $throwable) => $throwable instanceof HttpClientException
    )];
}

使用 Redis 对异常进行限流

如果你正在使用 Redis，可以使用 Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis 中间件，该中间件针对 Redis 进行了优化，比基础异常限流中间件更高效：

use Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis;

public function middleware(): array
{
    return [new ThrottlesExceptionsWithRedis(10, 10 * 60)];
}

可以使用 connection 方法来指定该中间件应使用的 Redis 连接：

return [(new ThrottlesExceptionsWithRedis(10, 10 * 60))->connection('limiter')];

释放作业

Release 中间件允许将作业释放回队列而不执行它。如果给定条件计算结果为 true，Release::when 方法将释放作业；而如果条件计算结果为 false，Release::unless 方法将释放作业：

use Illuminate\Queue\Middleware\Release;

/**
 * 获取作业应通过的中间件。
 */
public function middleware(): array
{
    return [
        Release::when($condition, releaseAfter: 60),
    ];
}

将作业释放回队列仍会增加作业的总尝试次数。你可能希望据此调整作业类中的 Tries 和 MaxExceptions 属性。

你也可以向 when 和 unless 方法传递 Closure，以便进行更复杂的条件评估：

use Illuminate\Queue\Middleware\Release;

/**
 * 获取作业应通过的中间件。
 */
public function middleware(): array
{
    return [
        Release::when(function (): bool {
            return ! $this->order->isPaid();
        }, releaseAfter: 60),
    ];
}

跳过作业

Skip 中间件允许指定作业应被跳过/删除，而无需修改作业的逻辑。如果给定条件计算结果为 true，Skip::when 方法将删除作业；而如果条件计算结果为 false，Skip::unless 方法将删除作业：

use Illuminate\Queue\Middleware\Skip;

/**
 * 获取作业应通过的中间件。
 */
public function middleware(): array
{
    return [
        Skip::when($condition),
    ];
}

你也可以向 when 和 unless 方法传递 Closure，以便进行更复杂的条件评估：

use Illuminate\Queue\Middleware\Skip;

/**
 * 获取作业应通过的中间件。
 */
public function middleware(): array
{
    return [
        Skip::when(function (): bool {
            return $this->shouldSkip();
        }),
    ];
}

分发作业

编写好作业类后，你可以使用作业自身的 dispatch 方法来分发它。传递给 dispatch 方法的参数将传入作业的构造函数：

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 存储新的播客。
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // ...

        ProcessPodcast::dispatch($podcast);

        return redirect('/podcasts');
    }
}

如果你想有条件地分发作业，可以使用 dispatchIf 和 dispatchUnless 方法：

ProcessPodcast::dispatchIf($accountActive, $podcast);

ProcessPodcast::dispatchUnless($accountSuspended, $podcast);

在新的 Laravel 应用程序中，database 连接被定义为默认队列。你可以通过修改应用程序 .env 文件中的 QUEUE_CONNECTION 环境变量，指定其他默认队列连接。

延迟分发

如果你想指定作业不应立即可供队列工作进程处理，可以在分发作业时使用 delay 方法。例如，我们指定作业在分发 10 分钟后才可供处理：

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 存储新的播客。
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // ...

        ProcessPodcast::dispatch($podcast)
            ->delay(now()->plus(minutes: 10));

        return redirect('/podcasts');
    }
}

在某些情况下，作业可能配置了默认延迟。如果你需要绕过此延迟并立即分发作业进行处理，可以使用 withoutDelay 方法：

ProcessPodcast::dispatch($podcast)->withoutDelay();

Amazon SQS 队列服务的最大延迟时间为 15 分钟。

同步分发

如果你想立即（同步）分发作业，可以使用 dispatchSync 方法。使用此方法时，作业不会被加入队列，而是在当前进程中立即执行：

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 存储新的播客。
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Create podcast...

        ProcessPodcast::dispatchSync($podcast);

        return redirect('/podcasts');
    }
}

延迟同步分发

使用延迟同步分发，你可以分发作业以便在当前进程中处理，但在 HTTP 响应已发送给用户之后。这允许你同步处理“已入队”的作业，而不会减慢用户的应用程序体验。要延迟执行同步作业，可将作业分发到 deferred 连接：

RecordDelivery::dispatch($order)->onConnection('deferred');

deferred 连接也可作为默认故障转移队列。

同样，background 连接在将 HTTP 响应发送给用户后处理作业；然而，作业会在单独启动的 PHP 进程中处理，从而让 PHP-FPM / 应用程序工作进程可用来处理另一个传入的 HTTP 请求：

RecordDelivery::dispatch($order)->onConnection('background');

批量分发

如果你需要一次性分发许多独立作业，且不需要批处理跟踪或回调，可以使用 Bus 外观的 bulk 方法。Laravel 会根据配置的队列连接和队列名称对作业进行分组，并将每组作业批量推送到相应的队列：

use App\Jobs\ProcessUser;
use Illuminate\Support\Facades\Bus;

Bus::bulk(
    $users->map(fn ($user) => new ProcessUser($user))
);

分发前准备作业

如果作业在被推送到队列之前需要准备或检查其状态，该作业可以实现 Illuminate\Contracts\Queue\PreparesForDispatch 接口。Laravel 会在分发作业前调用作业的 prepareForDispatch 方法。如果此方法返回 false，作业将不会被分发：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\PreparesForDispatch;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\Cache;

class SyncPodcasts implements PreparesForDispatch, ShouldQueue
{
    use Queueable;

    /**
     * 创建新的作业实例。
     */
    public function __construct(
        public array $podcastIds,
    ) {}

    /**
     * 在分发前准备作业。
     */
    public function prepareForDispatch(): bool
    {
        return collect($this->podcastIds)
            ->reject(fn (int $id) => Cache::has("podcast-syncing:{$id}"))
            ->isNotEmpty();
    }
}

作业与数据库事务

虽然在数据库事务中分发任务完全可行，但需格外留意，以确保任务确实能够成功执行。在事务内分发任务时，任务有可能在父事务提交前就被工作进程处理。这种情况下，你在数据库事务中对模型或数据库记录所做的任何更新，可能尚未反映到数据库里。此外，在事务中创建的任何模型或数据库记录，在数据库中可能并不存在。

好在 Laravel 提供了几种方法来规避这个问题。首先，你可以在队列连接的配置数组中设置 after_commit 连接选项：

'redis' => [
    'driver' => 'redis',
    // ...
    'after_commit' => true,
],

当 after_commit 选项为 true 时，你可以在数据库事务中分发任务；不过，Laravel 会等到已开启的父数据库事务提交后，才真正分发任务。当然，如果当前没有开启任何数据库事务，任务会立即被分发。

若事务因执行期间抛出异常而回滚，在该事务中分发的任务将会被丢弃。

将 after_commit 配置选项设为 true，也会让所有排队的事件监听器、邮件对象、通知以及广播事件，在所有已开启的数据库事务提交后再进行分发。

内联指定提交后分发行为

如果未将 after_commit 队列连接配置选项设为 true，你仍可指定某个具体任务应当在所有已开启的数据库事务提交后再分发。为此，你可以在分发操作上链式调用 afterCommit 方法：

use App\Jobs\ProcessPodcast;

ProcessPodcast::dispatch($podcast)->afterCommit();

同理，如果 after_commit 配置选项设为 true，你可以指定某个具体任务立即分发，无需等待任何已开启的数据库事务提交：

ProcessPodcast::dispatch($podcast)->beforeCommit();

任务链

任务链可让你指定一组排队任务，这些任务会在主任务执行成功后按顺序运行。若序列中某个任务失败，其余任务将不会执行。若要执行排队任务链，可以使用 Bus 门面提供的 chain 方法。Laravel 的命令总线是排队任务分发所依托的底层组件：

use App\Jobs\OptimizePodcast;
use App\Jobs\ProcessPodcast;
use App\Jobs\ReleasePodcast;
use Illuminate\Support\Facades\Bus;

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->dispatch();

除了对任务类实例进行链式编排，你也可以对闭包进行链式编排：

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    function () {
        Podcast::update(/* ... */);
    },
])->dispatch();

在任务内使用 $this->delete() 方法删除任务，不会阻止链中后续任务被处理。只有当链中某个任务失败时，任务链才会停止执行。

任务链的连接与队列

若想指定链中任务所使用的连接和队列，可以使用 onConnection 和 onQueue 方法。这些方法用于指定队列连接和队列名称，除非排队任务被显式分配了不同的连接/队列：

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->onConnection('redis')->onQueue('podcasts')->dispatch();

向任务链添加任务

有时，你可能需要从链中的某个任务内部，将一个任务添加到现有任务链的头部或尾部。你可以使用 prependToChain 和 appendToChain 方法来实现：

/**
 * Execute the job.
 */
public function handle(): void
{
    // ...

    // Prepend to the current chain, run job immediately after current job...
    $this->prependToChain(new TranscribePodcast);

    // Append to the current chain, run job at end of chain...
    $this->appendToChain(new TranscribePodcast);
}

任务链失败处理

编排任务链时，可以使用 catch 方法指定一个闭包，当链中某个任务失败时应调用该闭包。传入的回调会接收导致任务失败的 Throwable 实例：

use Illuminate\Support\Facades\Bus;
use Throwable;

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->catch(function (Throwable $e) {
    // A job within the chain has failed...
})->dispatch();

由于任务链回调会被序列化，并在稍后由 Laravel 队列执行，因此不应在任务链回调中使用 $this 变量。

自定义队列与连接

分发到指定队列

通过将任务推送到不同的队列，你可以对排队任务进行“分类”，甚至优先分配不同数量的工作进程处理各类队列。需注意，这并不会把任务推送到队列配置文件中定义的不同队列“连接”，而是仅推送到单个连接内的特定队列。若要指定队列，可在分发任务时使用 onQueue 方法：

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Create podcast...

        ProcessPodcast::dispatch($podcast)->onQueue('processing');

        return redirect('/podcasts');
    }
}

或者，你也可以在任务的构造函数中调用 onQueue 方法，来指定任务的队列：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct()
    {
        $this->onQueue('processing');
    }
}

分发到指定连接

如果你的应用与多个队列连接交互，你可以使用 onConnection 方法指定要将任务推送到哪个连接：

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Create podcast...

        ProcessPodcast::dispatch($podcast)->onConnection('sqs');

        return redirect('/podcasts');
    }
}

你可以将 onConnection 和 onQueue 方法链式调用，来指定任务的连接与队列：

ProcessPodcast::dispatch($podcast)
    ->onConnection('sqs')
    ->onQueue('processing');

或者，你也可以在任务的构造函数中调用 onConnection 方法，来指定任务的连接：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct()
    {
        $this->onConnection('sqs');
    }
}

队列路由

你可以使用 Queue 门面的 route 方法，为特定任务类定义默认的连接和队列。当你想确保某些任务始终使用特定队列，又不想在任务上显式指定连接或队列时，这种方式十分有用。

除了为特定任务类配置路由，你也可以向 route 方法传入接口、trait 或父类。这样做后，任何实现了该接口、使用了该 trait 或继承了该父类的任务，都会自动使用所配置的连接和队列。

通常，你应该在服务提供者的 boot 方法中调用 route 方法：

use App\Concerns\RequiresVideo;
use App\Jobs\ProcessPodcast;
use App\Jobs\ProcessVideo;
use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
use Illuminate\Support\Facades\Queue;

/**
 * Bootstrap any application services.
 */
public function boot(): void
{
    Queue::route(ProcessPodcast::class, connection: 'redis', queue: 'podcasts');
    Queue::route(RequiresVideo::class, queue: 'video');
    Queue::route(ShouldBroadcast::class, queue: 'events');
}

当指定连接时未指定队列，任务将被发送到默认队列：

Queue::route(ProcessPodcast::class, connection: 'redis');

也可以通过向 route 方法传递数组，一次性路由多个任务类：

Queue::route([
    ProcessPodcast::class => ['redis', 'podcasts'], // 连接和队列
    ProcessVideo::class => 'videos', // 仅队列（使用默认连接）
]);

队列路由仍可由任务按单个任务进行覆盖。

可以使用 forward 方法将任务从一个队列转发到另一个队列和/或连接。当需要更改队列基础设施而不修改单个任务或调度位置时，此方法非常有用：

Queue::forward('reports', 'reports.fifo', 'sqs');
Queue::forward('payments', connection: 'sqs');
Queue::forward('updates', 'notifications');

也可以通过传递数组，一次性转发多个队列：

Queue::forward([
    'reports' => 'reports.fifo',
    'emails' => 'emails.fifo',
], connection: 'sqs');

在任务上配置的显式连接优先级高于被转发的连接。

指定最大任务尝试次数 / 超时值

最大尝试次数

任务尝试是 Laravel 队列系统的核心概念，支撑了许多高级功能。尽管它们最初看起来可能令人困惑，但在修改默认配置之前，理解其工作原理非常重要。

当任务被调度时，它会被推送到队列中。随后，工作进程会获取该任务并尝试执行。这就是一次任务尝试。

然而，一次尝试不一定意味着任务的 handle 方法已被执行。尝试还可能通过以下几种方式被“消耗”：

• 任务在执行过程中遇到未处理的异常。
• 任务通过 $this->release() 被手动释放回队列。
• 像 WithoutOverlapping 或 RateLimited 这样的中间件未能获取锁，并释放了任务。
• 任务超时。
• 任务的 handle 方法运行完成且未抛出异常。

你可能不希望无限次地尝试某个任务。因此，Laravel 提供了多种方式来指定一个任务可以被尝试的次数或时长。

默认情况下，Laravel 只会尝试执行任务一次。如果你的任务使用了像 WithoutOverlapping 或 RateLimited 这样的中间件，或者你手动释放了任务，那么你很可能需要通过 tries 选项增加允许的尝试次数。

指定任务最大尝试次数的一种方式是使用 Artisan 命令行上的 --tries 开关。这将适用于工作进程处理的所有任务，除非被处理的任务自行指定了可尝试的次数：

php artisan queue:work --tries=3

如果任务超出了最大尝试次数，它将被视为“失败”任务。有关处理失败任务的更多信息，请参阅失败任务文档。如果向 queue:work 命令提供 --tries=0，任务将被无限重试。

你也可以通过在任务类上使用 Tries 特性来定义任务的最大尝试次数，这种方式更加精细。如果在任务上指定了最大尝试次数，其优先级将高于命令行中提供的 --tries 值：

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Tries;

#[Tries(5)]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

如果需要对特定任务的最大尝试次数进行动态控制，可以在任务中定义一个 tries 方法：

/**
 * 确定任务可以被尝试的次数。
 */
public function tries(): int
{
    return 5;
}

基于时间的尝试

作为定义任务失败前可尝试次数的替代方案，你可以定义一个时间点，在此时间之后任务不再被尝试。这样，任务可以在给定时间范围内被尝试任意多次。要定义任务不再被尝试的时间，请在任务类中添加一个 retryUntil 方法。该方法应返回一个 DateTime 实例：

use DateTime;

/**
 * 确定任务应超时的时间。
 */
public function retryUntil(): DateTime
{
    return now()->plus(minutes: 10);
}

如果同时定义了 retryUntil 和 tries，Laravel 会优先使用 retryUntil 方法。

你也可以在排队的事件监听器和排队的通知上定义 Tries 特性或 retryUntil 方法。

最大异常数

有时你可能希望指定任务可以被尝试多次，但如果重试是由给定数量的未处理异常触发的（而不是由 release 方法直接释放），则任务应失败。为此，你可以在任务类上使用 Tries 和 MaxExceptions 特性：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\MaxExceptions;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Support\Facades\Redis;

#[Tries(25)]
#[MaxExceptions(3)]
class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * 执行任务。
     */
    public function handle(): void
    {
        Redis::throttle('key')->allow(10)->every(60)->then(function () {
            // 获取到锁，处理播客...
        }, function () {
            // 无法获取锁...
            return $this->release(10);
        });
    }
}

在此示例中，如果应用无法获取 Redis 锁，任务将被释放 10 秒，并继续重试最多 25 次。但是，如果任务抛出三个未处理的异常，任务将失败。

默认情况下，因工作进程崩溃或被终止（例如内存耗尽）而导致的一次尝试，不会计入任务的最大异常数。如果你希望这些尝试被计为异常，可以在任务类中添加 CountCrashesAsExceptions 特性：

use Illuminate\Queue\Attributes\CountCrashesAsExceptions;
use Illuminate\Queue\Attributes\MaxExceptions;
use Illuminate\Queue\Attributes\Tries;

#[Tries(25)]
#[MaxExceptions(3)]
#[CountCrashesAsExceptions]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

当存在此特性时，工作进程在处理任务期间会在应用的缓存中存储一个标记。如果下一次尝试该任务时标记仍然存在，则之前的尝试将被计为一次异常。

通过异常停止重试

有时，异常表明排队的任务应立即失败，而不是被释放以进行下一次尝试。你可以在应用的 bootstrap/app.php 文件中使用 dontRetry 异常方法，配置应停止任务重试的异常类型：

use App\Exceptions\InvalidPodcastSourceException;
use Illuminate\Foundation\Configuration\Exceptions;

->withExceptions(function (Exceptions $exceptions): void {
    $exceptions->dontRetry([
        InvalidPodcastSourceException::class,
    ]);
})

如果你需要对何时停止重试进行更精细的控制，可以向 dontRetryWhen 方法提供一个闭包。当闭包返回 true 时，任务将被标记为失败，且不会重试：

use App\Exceptions\PodcastProcessingException;
use Illuminate\Foundation\Configuration\Exceptions;

->withExceptions(function (Exceptions $exceptions): void {
    $exceptions->dontRetryWhen(function (PodcastProcessingException $e) {
        return $e->reason() === 'Subscription expired';
    });
})

超时

通常，你可以大致预估排队任务所需的执行时间。因此，Laravel 允许你指定一个“超时”值。默认情况下，超时值为 60 秒。如果一个任务的处理时间超过了超时值指定的秒数，处理该任务的工作进程将因错误退出。通常，服务器上配置的进程管理器会自动重启该工作进程。

任务可以运行的最大秒数可以通过 Artisan 命令行上的 --timeout 开关来指定：

php artisan queue:work --timeout=30

如果任务因持续超时而超出最大尝试次数，它将被标记为失败。

你还可以使用任务类上的 Timeout 特性来定义任务允许运行的最大秒数。如果在任务上指定了超时，其优先级将高于命令行中指定的任何超时值：

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Timeout;

#[Timeout(120)]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

有时，像套接字或出站 HTTP 连接这类 I/O 阻塞进程可能不会遵循你指定的超时时间。因此，在使用这些功能时，你应始终尝试通过它们的 API 来指定超时时间。例如，在使用 Guzzle 时，你应始终指定连接和请求的超时值。

若要指定任务超时，必须安装 PCNTL PHP 扩展。此外，任务的“超时”值应始终小于其“重试间隔”值。否则，任务可能在尚未完成执行或尚未超时之前就被重新尝试。当使用 --once 选项调用 queue:work 命令时，--timeout 选项不起作用。

超时时失败

如果你希望在任务超时时将其标记为失败，可以在任务类上使用 FailOnTimeout 特性：

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\FailOnTimeout;

#[FailOnTimeout]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

默认情况下，当任务超时时，它会消耗一次尝试并重新放回队列（若允许重试）。不过，如果你将任务配置为在超时时失败，无论设置的尝试次数是多少，该任务都不会被重试。

SQS FIFO 与公平队列

Laravel 支持 Amazon SQS FIFO（先进先出） 和 公平 队列。FIFO 队列能让你按任务发送的确切顺序处理任务，并通过消息去重确保任务仅被处理一次。

FIFO 队列需要一个消息组 ID，用于确定哪些任务可并行处理。具有相同组 ID 的任务按顺序处理，而具有不同组 ID 的消息则可并发处理。

Laravel 提供了流畅的 onGroup 方法，可在分发任务时指定消息组 ID：

ProcessOrder::dispatch($order)
    ->onGroup("customer-{$order->customer_id}");

如果你将任务分发到 SQS FIFO 队列时未指定消息组，Laravel 会将队列名称用作消息组 ID。

SQS FIFO 队列支持消息去重，以确保任务仅被处理一次。你可以在任务类中实现 deduplicationId 方法，以提供自定义的去重 ID：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessSubscriptionRenewal implements ShouldQueue
{
    use Queueable;

    // ...

    /**
     * 获取任务的去重 ID。
     */
    public function deduplicationId(): string
    {
        return "renewal-{$this->subscription->id}";
    }
}

公平队列

若你使用的是 SQS 标准队列，设置消息组即可启用公平队列。换言之，一旦你分配了组，SQS 就会用这些组在不同租户/工作负载之间维持公平投递，无需进行额外的 Laravel 配置。

除了在分发时调用 onGroup，你也可以直接在任务中定义 messageGroup 方法：

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessOrder implements ShouldQueue
{
    use Queueable;

    // ...

    /**
     * 获取任务的消息组。
     */
    public function messageGroup(): string
    {
        return "customer-{$this->order->customer_id}";
    }
}

FIFO 监听器、邮件与通知

使用 FIFO 队列时，你还需要在监听器、邮件和通知上定义消息组。或者，你可以将这些对象的队列实例分发到非 FIFO 队列。

要为 队列化事件监听器 定义消息组，请在监听器上定义 messageGroup 方法。你还可以选择定义一个 deduplicator 方法，该方法接收事件，并应返回一个用于生成去重 ID 的闭包：

<?php

namespace App\Listeners;

use App\Events\OrderShipped;
use Closure;

class SendShipmentNotification
{
    // ...

    /**
     * 获取任务的消息组。
     */
    public function messageGroup(): string
    {
        return 'shipments';
    }

    /**
     * 获取任务的去重器。
     */
    public function deduplicator(OrderShipped $event): Closure
    {
        return fn () => "shipment-notification-{$event->order->id}";
    }
}

若要发送将排入 FIFO 队列的 邮件，在发送通知时应调用 onGroup 方法，并可选择调用 withDeduplicator 方法：

use App\Mail\InvoicePaid;
use Illuminate\Support\Facades\Mail;

$invoicePaid = (new InvoicePaid($invoice))
    ->onGroup('invoices')
    ->withDeduplicator(fn () => 'invoices-'.$invoice->id);

Mail::to($request->user())->send($invoicePaid);

若要发送将排入 FIFO 队列的 通知，在发送通知时应调用 onGroup 方法，并可选择调用 withDeduplicator 方法：

use App\Notifications\InvoicePaid;

$invoicePaid = (new InvoicePaid($invoice))
    ->onGroup('invoices')
    ->withDeduplicator(fn () => 'invoices-'.$invoice->id);

$user->notify($invoicePaid);

队列故障转移

failover 队列驱动在向队列推送任务时提供自动故障转移功能。若 failover 配置中的主队列连接因任何原因失败，Laravel 会自动尝试将任务推送到列表中的下一个已配置连接。这对于在对队列可靠性要求极高的生产环境中保障高可用性尤为有用。

要配置故障转移队列连接，请指定 failover 驱动，并提供一个按顺序尝试的连接名称数组。默认情况下，Laravel 会在应用的 config/queue.php 配置文件中包含一个故障转移配置示例：

'failover' => [
    'driver' => 'failover',
    'connections' => [
        'redis',
        'database',
        'sync',
    ],
],

配置好使用 failover 驱动的连接后，你需要在应用的 .env 文件中将故障转移连接设置为默认队列连接，才能使用故障转移功能：

QUEUE_CONNECTION=failover

接下来，请为故障转移连接列表中的每个连接至少启动一个工作进程：

php artisan queue:work redis
php artisan queue:work database

对于使用 sync、background 或 deferred 队列驱动的连接，无需运行工作进程，因为这些驱动会在当前 PHP 进程中处理任务。

当队列连接操作失败并触发故障转移时，Laravel 会派发 Illuminate\Queue\Events\QueueFailedOver 事件，以便你报告或记录队列连接已失败。

如果你使用 Laravel Horizon，请记住 Horizon 仅管理 Redis 队列。若你的故障转移列表包含 database，则应在运行 Horizon 的同时运行常规的 php artisan queue:work database 进程。

错误处理

若在处理任务时抛出异常，任务会自动被重新放回队列以便再次尝试。任务会持续被重新放回队列，直到达到应用所允许的最大尝试次数。最大尝试次数由 queue:work Artisan 命令中使用的 --tries 选项定义。也可以在任务类自身上定义最大尝试次数。有关运行队列工作进程的更多信息，请见下文。

手动释放任务

有时你可能希望将任务手动放回队列，以便稍后再次尝试。你可以通过调用 release 方法来实现：

/**
 * 执行任务。
 */
public function handle(): void
{
    // ...

    $this->release();
}

默认情况下，release 方法会将任务放回队列以立即处理。不过，你可以向 release 方法传入整数或日期实例，指示队列在经过指定秒数之前不要让该任务可被处理：

$this->release(10);

$this->release(now()->plus(seconds: 10));

手动让任务失败

有时你可能需要手动将任务标记为“失败”。为此，你可以调用 fail 方法：

/**
 * 执行任务。
 */
public function handle(): void
{
    // ...

    $this->fail();
}

如果你想由于已捕获的异常而将任务标记为失败，可以将该异常传递给 fail 方法。或者，为方便起见，你也可以传入一个字符串错误消息，该方法会为你将其转换为异常：

$this->fail($exception);

$this->fail('Something went wrong.');

如需了解有关失败任务的更多信息，请参阅处理任务失败的文档。

在特定异常时让任务失败

FailOnException 任务中间件允许在抛出特定异常时提前终止重试。这样可以在遇到诸如外部 API 错误之类的临时性异常时继续重试，而在遇到诸如用户权限被撤销之类的持续性异常时永久使任务失败：

<?php

namespace App\Jobs;

use App\Models\User;
use Illuminate\Auth\Access\AuthorizationException;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Queue\Middleware\FailOnException;
use Illuminate\Support\Facades\Http;

#[Tries(3)]
class SyncChatHistory implements ShouldQueue
{
    use Queueable;

    /**
     * 创建新的任务实例。
     */
    public function __construct(
        public User $user,
    ) {}

    /**
     * 执行任务。
     */
    public function handle(): void
    {
        $this->user->authorize('sync-chat-history');

        $response = Http::throw()->get(
            "https://chat.laravel.test/?user={$this->user->uuid}"
        );

        // ...
    }

    /**
     * 获取该任务应经过的中间件。
     */
    public function middleware(): array
    {
        return [
            new FailOnException([AuthorizationException::class])
        ];
    }
}

任务批处理

Laravel 的任务批处理功能允许你轻松地并行执行一组任务，并在该批任务执行完成后执行某些操作。

在开始之前，你需要创建一个数据库迁移以构建一张表，用于存储有关任务批的元信息，例如其完成百分比。该迁移可通过 make:queue-batches-table Artisan 命令生成：

php artisan make:queue-batches-table

php artisan migrate

定义可批处理的任务

要定义一个可批处理的任务，你应按常规创建一个可入队的任务；不过，你需要在任务类中添加 Illuminate\Bus\Batchable trait。该 trait 提供了对 batch 方法的访问，可用于获取该任务当前正在执行的批次：

<?php

namespace App\Jobs;

use Illuminate\Bus\Batchable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ImportCsv implements ShouldQueue
{
    use Batchable, Queueable;

    /**
     * 执行任务。
     */
    public function handle(): void
    {
        if ($this->batch()->cancelled()) {
            // 判断该批次是否已被取消...

            return;
        }

        // 导入 CSV 文件的一部分...
    }
}

派发批次

要派发一批任务，应使用 Bus 门面的 batch 方法。当然，批处理只有在结合完成回调时才真正有用。因此，你可以使用 then、catch 和 finally 方法来为该批次定义完成回调。这些回调在被调用时都会接收到一个 Illuminate\Bus\Batch 实例。

当运行多个队列工作进程时，批次中的任务会并行处理。因此，任务完成的顺序可能与它们被添加到批次的顺序不同。关于如何按顺序运行一系列任务，请参阅我们关于任务链与批次的文档。

在此示例中，我们将想象正在入队一个任务批次，其中每个任务都会处理 CSV 文件中指定数量的行：

use App\Jobs\ImportCsv;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;
use Throwable;

$batch = Bus::batch([
    new ImportCsv(1, 100),
    new ImportCsv(101, 200),
    new ImportCsv(201, 300),
    new ImportCsv(301, 400),
    new ImportCsv(401, 500),
])->before(function (Batch $batch) {
    // 批次已创建，但尚未添加任何任务...
})->progress(function (Batch $batch) {
    // 单个任务已成功完成...
})->then(function (Batch $batch) {
    // 所有任务均已成功完成...
})->catch(function (Batch $batch, Throwable $e) {
    // 检测到批次任务失败...
})->finally(function (Batch $batch) {
    // 批次已执行完毕...
})->dispatch();

return $batch->id;

该批次的 ID（可通过 $batch->id 属性访问）可用于在批次派发之后向 Laravel 命令总线查询有关该批次的信息。

由于批次回调会被序列化，并在稍后由 Laravel 队列执行，因此不应在回调中使用 $this 变量。此外，由于被批处理的任务会被包裹在数据库事务中，因此不应在任务中执行会触发隐式提交的数据库语句。

为批次命名

如果为批次命名，部分工具（例如Laravel Horizon 与 Laravel Telescope）可能会为批次提供更易于使用的调试信息。要为批次指定一个任意名称，你可以在定义批次时调用 name 方法：

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // 所有任务均已成功完成...
})->name('Import CSV')->dispatch();

批次的连接与队列

如果你希望为被批处理的任务指定应使用的连接与队列，可以使用 onConnection 和 onQueue 方法。所有被批处理的任务必须在同一个连接与队列中执行：

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // 所有任务均已成功完成...
})->onConnection('redis')->onQueue('imports')->dispatch();

链与批次

你可以将一组任务链放在一个数组中，从而在批次内定义这些链式任务。例如，我们可以并行执行两条任务链，并在两条任务链都处理完成后执行一个回调：

use App\Jobs\ReleasePodcast;
use App\Jobs\SendPodcastReleaseNotification;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;

Bus::batch([
    [
        new ReleasePodcast(1),
        new SendPodcastReleaseNotification(1),
    ],
    [
        new ReleasePodcast(2),
        new SendPodcastReleaseNotification(2),
    ],
])->then(function (Batch $batch) {
    // 所有任务均已成功完成...
})->dispatch();

相反，你也可以通过在链中定义批次来在链内运行任务批次。例如，你可以先运行一个批次来发布多个播客，然后运行一个批次来发送发布通知：

use App\Jobs\FlushPodcastCache;
use App\Jobs\ReleasePodcast;
use App\Jobs\SendPodcastReleaseNotification;
use Illuminate\Support\Facades\Bus;

Bus::chain([
    new FlushPodcastCache,
    Bus::batch([
        new ReleasePodcast(1),
        new ReleasePodcast(2),
    ]),
    Bus::batch([
        new SendPodcastReleaseNotification(1),
        new SendPodcastReleaseNotification(2),
    ]),
])->dispatch();

向批次中添加任务

有时，从某个属于批次的任务内部向该批次中添加额外任务会很有用。当你需要批处理成千上万个任务、而在 Web 请求期间全部派发耗时过长时，这种模式会很有用。因此，你可以选择先派发一批初始的“加载器”任务，由它们为该批次补充更多任务：

$batch = Bus::batch([
    new LoadImportBatch,
    new LoadImportBatch,
    new LoadImportBatch,
])->then(function (Batch $batch) {
    // 所有任务均已成功完成...
})->name('Import Contacts')->dispatch();

在此示例中，我们将使用 LoadImportBatch 任务为该批次补充额外任务。为实现这一点，可以使用批次实例上的 add 方法；你可以通过任务的 batch 方法获取该批次实例：

use App\Jobs\ImportContacts;
use Illuminate\Support\Collection;

/**
 * 执行任务。
 */
public function handle(): void
{
    if ($this->batch()->cancelled()) {
        return;
    }

    $this->batch()->add(Collection::times(1000, function () {
        return new ImportContacts;
    }));
}

你只能从属于同一批次的任务内部向该批次中添加任务。

检查批次

提供给批次完成回调的 Illuminate\Bus\Batch 实例具有多种属性和方法，可帮助你与给定的任务批次进行交互并对其进行检查：

// 批次的 UUID……
$batch->id;

// 批次的名称（如果适用）……
$batch->name;

// 分配给该批次的任务数量……
$batch->totalJobs;

// 尚未被队列处理的任务数量……
$batch->pendingJobs;

// 已失败的任务数量……
$batch->failedJobs;

// 截至目前已处理的任务数量……
$batch->processedJobs();

// 批次的完成百分比（0-100）……
$batch->progress();

// 指示批次是否已执行完成……
$batch->finished();

// 取消批次的执行……
$batch->cancel();

// 指示批次是否已被取消……
$batch->cancelled();

从路由返回批次

所有 Illuminate\Bus\Batch 实例均可进行 JSON 序列化，这意味着你可以直接在应用的某个路由中返回它们，从而获取包含批次信息（包括其完成进度）的 JSON 数据。这让你在应用的用户界面中显示批次完成进度信息变得更加方便。

若要通过 ID 检索批次，你可以使用 Bus 门面的 findBatch 方法：

use Illuminate\Support\Facades\Bus;
use Illuminate\Support\Facades\Route;

Route::get('/batch/{batchId}', function (string $batchId) {
    return Bus::findBatch($batchId);
});

取消批次

有时你可能需要取消某个批次的执行。这可以通过在 Illuminate\Bus\Batch 实例上调用 cancel 方法来实现：

/**
 * 执行任务。
 */
public function handle(): void
{
    if ($this->user->exceedsImportLimit()) {
        $this->batch()->cancel();

        return;
    }

    if ($this->batch()->cancelled()) {
        return;
    }
}

正如在前面的示例中你可能注意到的那样，分批任务通常应在继续执行之前判断其对应的批次是否已被取消。不过，为了方便起见，你可以为任务分配 SkipIfBatchCancelled 中间件。顾名思义，若其对应的批次已被取消，该中间件会指示 Laravel 不处理此任务：

use Illuminate\Queue\Middleware\SkipIfBatchCancelled;

/**
 * 获取任务应经过的中间件。
 */
public function middleware(): array
{
    return [new SkipIfBatchCancelled];
}

批次失败

当某个分批任务失败时，catch 回调（若已指定）将被触发。该回调仅会在批次内首个失败的任务上被调用。

允许失败

当批次中的某个任务失败时，Laravel 会自动将该批次标记为“已取消”。如果你希望在任务失败时不自动将批次标记为已取消，可以禁用此行为。你可以在分发批次时调用 allowFailures 方法来实现这一点：

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // 所有任务均已成功完成……
})->allowFailures()->dispatch();

你也可以选择性地为 allowFailures 方法提供一个闭包，该闭包会在每个任务失败时执行：

$batch = Bus::batch([
    // ...
])->allowFailures(function (Batch $batch, $exception) {
    // 处理单个任务失败的情况……
})->dispatch();

重试失败的批次任务

为方便起见，Laravel 提供了 queue:retry-batch Artisan 命令，让你可以轻松重试指定批次中所有失败的任务。该命令接受需要重试失败任务的批次的 UUID：

php artisan queue:retry-batch 32dbc76c-4f82-4749-b610-a639fe0099b5

清理批次

如果不进行清理，job_batches 表可能会很快累积大量记录。为了缓解这一问题，你应该按日调度 queue:prune-batches Artisan 命令运行：

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches')->daily();

默认情况下，所有超过 24 小时的已完成批次都会被清理。你可以在调用该命令时使用 hours 选项来指定批次数据的保留时长。例如，以下命令会删除所有在 48 小时前已完成的批次：

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches --hours=48')->daily();

有时，你的 job_batches 表中可能会累积从未成功完成的批次记录，例如某批次中的某个任务失败且该任务从未被成功重试的情况。你可以使用 unfinished 选项，指示 queue:prune-batches 命令清理这些未完成的批次记录：

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches --hours=48 --unfinished=72')->daily();

同样，你的 job_batches 表也可能累积已取消批次的记录。你可以使用 cancelled 选项，指示 queue:prune-batches 命令清理这些已取消的批次记录：

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches --hours=48 --cancelled=72')->daily();

将批次存储在 DynamoDB 中

Laravel 也支持将批次元信息存储在 DynamoDB 中，而非关系型数据库中。不过，你需要手动创建一个 DynamoDB 表来存储所有的批次记录。

通常，该表应命名为 job_batches，但你应根据应用 queue 配置文件中 queue.batching.table 配置项的值来命名该表。

DynamoDB 批次表配置

job_batches 表应包含一个名为 application 的字符串类型主分区键，以及一个名为 id 的字符串类型主排序键。键中的 application 部分将包含你的应用名称，该名称由应用 app 配置文件中 name 配置项的值定义。由于应用名称是 DynamoDB 表键的一部分，因此你可以使用同一张表来存储多个 Laravel 应用的任务批次。

此外，如果你想利用自动批次清理功能，可以为该表定义 ttl 属性。

DynamoDB 配置

接下来，安装 AWS SDK，以便你的 Laravel 应用能够与 Amazon DynamoDB 进行通信：

composer require aws/aws-sdk-php

然后，将 queue.batching.driver 配置项的值设置为 dynamodb。此外，你应该在 batching 配置数组中定义 key、secret 和 region 配置项。这些配置项将用于与 AWS 进行身份验证。当使用 dynamodb 驱动时，无需配置 queue.batching.database 配置项：

'batching' => [
    'driver' => env('QUEUE_BATCHING_DRIVER', 'dynamodb'),
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'table' => 'job_batches',
],

在 DynamoDB 中清理批次

当使用 DynamoDB 存储任务批次信息时，通常用于清理关系型数据库中批次的标准清理命令将无法生效。此时，你可以利用 DynamoDB 原生的 TTL 功能 来自动移除旧批次的记录。

如果你在创建 DynamoDB 表时定义了 ttl 属性，就可以通过配置参数指示 Laravel 如何清理批次记录。其中，queue.batching.ttl_attribute 配置项用于定义存储 TTL 的属性名称，而 queue.batching.ttl 配置项则用于定义从最后一次更新记录起经过多少秒后，可将该批次记录从 DynamoDB 表中移除：

'batching' => [
    'driver' => env('QUEUE_BATCHING_DRIVER', 'dynamodb'),
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'table' => 'job_batches',
    'ttl_attribute' => 'ttl',
    'ttl' => 60 * 60 * 24 * 7, // 7 天……
],

对闭包进行排队

除了将任务类分发到队列之外，你也可以分发闭包。这对于需要在当前请求周期之外执行的快速、简单的任务而言非常合适。当向队列分发闭包时，闭包的代码内容会经过加密签名，以确保其不会在传输过程中被篡改：

use App\Models\Podcast;

$podcast = Podcast::find(1);

dispatch(function () use ($podcast) {
    $podcast->publish();
});

若要为队列闭包指定一个名称，该名称可被队列报表仪表板使用，也会由 queue:work 命令显示，可以使用 name 方法：

dispatch(function () {
    // ...
})->name('Publish Podcast');

使用 catch 方法，你可以提供一个闭包，当队列闭包在耗尽队列中所有配置的重试次数后仍无法成功完成时，将执行该闭包：

use Throwable;

dispatch(function () use ($podcast) {
    $podcast->publish();
})->catch(function (Throwable $e) {
    // This job has failed...
});

由于 catch 回调会被序列化，并在之后由 Laravel 队列执行，因此你不应在使用 catch 回调时使用 $this 变量。

运行队列工作进程

queue:work 命令

Laravel 内置了一个 Artisan 命令，它会启动一个队列工作进程，并在新任务被推送到队列时对其进行处理。你可以使用 queue:work Artisan 命令来运行该工作进程。请注意，一旦 queue:work 命令启动，它将持续运行，直至你手动停止或关闭终端：

php artisan queue:work

若要让 queue:work 进程在后台永久运行，应使用 Supervisor 等进程监控工具，以确保队列工作进程不会停止运行。

调用 queue:work 命令时，如果你希望命令输出中包含已处理任务的 ID、连接名称和队列名称，可添加 -v 标志：

php artisan queue:work -v

请记住，队列工作进程是长驻进程，会将已启动的应用状态保存在内存中。因此，进程启动后将不会感知代码库中的变更。所以，在部署过程中，请务必重启队列工作进程。此外，还需注意，应用创建或修改的任何静态状态都不会在任务之间自动重置。

或者，你也可以运行 queue:listen 命令。使用 queue:listen 命令时，当你想重新加载更新后的代码或重置应用状态，无需手动重启工作进程；不过，该命令的效率远低于 queue:work 命令：

php artisan queue:listen

运行多个队列工作进程

若要为某个队列分配多个工作进程以并发处理任务，只需启动多个 queue:work 进程即可。你可以在当地通过终端中的多个标签页来实现，也可以在生产环境中通过进程管理器的配置设置来实现。使用 Supervisor 时，可以使用 numprocs 配置值。

指定连接和队列

你还可以指定工作进程应使用的队列连接。传递给 work 命令的连接名称，应与 config/queue.php 配置文件中定义的某个连接相对应：

php artisan queue:work redis

默认情况下，queue:work 命令只会处理指定连接上的默认队列中的任务。不过，你还可以对队列工作进程进行更细致的自定义，使其仅处理指定连接中的特定队列。例如，如果你所有的邮件任务都在 redis 队列连接的 emails 队列中进行处理，那么你可以执行以下命令，以启动一个仅处理该队列的工作进程：

php artisan queue:work redis --queue=emails

处理指定数量的任务

使用 --once 选项可以指示工作进程仅处理队列中的一个任务：

php artisan queue:work --once

使用 --max-jobs 选项可以指示工作进程处理指定数量的任务后退出。将该选项与 Supervisor 结合使用可能会很有用，这样在处理完指定数量的任务后，工作进程会自动重启，从而释放其可能累积的任何内存：

php artisan queue:work --max-jobs=1000

处理完所有队列任务后退出

使用 --stop-when-empty 选项可以指示工作进程处理完所有任务后优雅退出。如果你希望在队列为空后关闭 Docker 容器，那么在 Docker 容器内处理 Laravel 队列时，该选项会很有用：

php artisan queue:work --stop-when-empty

处理任务的时长（秒）

使用 --max-time 选项可以指示工作进程处理任务的时长，达到指定秒数后退出。将该选项与 Supervisor 结合使用可能会很有用，这样在处理一段时间的任务后，工作进程会自动重启，从而释放其可能累积的任何内存：

# Process jobs for one hour and then exit...
php artisan queue:work --max-time=3600

工作进程休眠时长

当队列中有可用任务时，工作进程会持续处理任务，且任务之间无延迟。不过，当没有可用任务时，sleep 选项会决定工作进程“休眠”的秒数。当然，在休眠期间，工作进程不会处理任何新任务：

php artisan queue:work --sleep=3

维护模式与队列

当应用处于维护模式时，队列任务将不会被处理。应用退出维护模式后，任务会恢复正常处理。

若要在开启维护模式的情况下，强制队列工作进程处理任务，可以使用 --force 选项：

php artisan queue:work --force

资源注意事项

--memory 选项用于指定工作进程的内存限制（以兆字节为单位），达到该限制时工作进程应退出。默认情况下，该限制为 128 兆字节。你也可以指定 PHP 配置的 memory_limit 的百分比：

php artisan queue:work --memory=60%

基于百分比的内存限制要求 PHP 的 memory_limit 为有限值。如果 PHP 的内存限制是无限制的（-1），则应使用以兆字节为单位的限制。像 Supervisor 这样的进程监控工具，可在工作进程退出后自动重启它。

守护进程式的队列工作进程在处理每个任务之前，不会“重新启动”框架。因此，你应在每个任务完成后释放所有占用大量内存的资源。例如，如果你正在使用图像处理功能，并使用GD 库，那么在完成图像处理后，应通过 imagedestroy 释放内存。

队列优先级

有时，你可能希望对队列的处理顺序设置优先级。例如，在 config/queue.php 配置文件中，你可以将 redis 连接的默认 queue 设置为 low。不过，偶尔你可能希望将任务推送到 high 优先级队列，如下所示：

dispatch((new Job)->onQueue('high'));

若要启动一个工作进程，使其先处理完 high 队列中的所有任务，再处理 low 队列中的任务，可以将以逗号分隔的队列名称列表传递给 work 命令：

php artisan queue:work --queue=high,low

队列工作进程与部署

由于队列工作进程是长驻进程，若不重启，它们不会感知代码的变更。因此，使用队列工作进程部署应用的最简单方法，就是在部署过程中重启工作进程。你可以通过执行 queue:restart 命令，优雅地重启所有工作进程：

php artisan queue:restart

该命令会指示所有队列工作进程在完成当前任务的处理后优雅退出，从而避免现有任务丢失。由于执行 queue:restart 命令后队列工作进程会退出，因此你应该运行像 Supervisor 这样的进程管理器，以便自动重启队列工作进程。

队列使用缓存来存储重启信号，因此在使用该功能之前，请确认应用的缓存驱动已正确配置。

响应工作进程信号

当队列工作进程在处理任务时收到终止信号（如 SIGQUIT、SIGTERM 或 SIGINT），工作进程会在退出前完成当前任务的处理。不过，在服务器或容器编排器停止进程之前，你的任务可能需要对信号作出响应。例如，一个长时间运行的导入任务，可能需要停止拉取新记录，并保存当前的进度。

若要在作业（job）内部响应工作进程信号，需实现 Illuminate\Contracts\Queue\Interruptible 接口，并在你的作业中定义 interrupted 方法。工作进程收到的信号编号会被传递给 interrupted 方法：

<?php

namespace App\Jobs;

use App\Models\Import;
use Illuminate\Contracts\Queue\Interruptible;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ImportProducts implements ShouldQueue, Interruptible
{
    use Queueable;

    protected bool $shouldStop = false;

    /**
     * 创建新的作业实例。
     */
    public function __construct(
        public Import $import,
    ) {}

    /**
     * 执行作业。
     */
    public function handle(): void
    {
        foreach ($this->import->pendingRows() as $row) {
            if ($this->shouldStop) {
                break;
            }

            // 导入产品行...
        }

        $this->import->saveProgress();
    }

    /**
     * 处理队列工作进程接收到的信号。
     */
    public function interrupted(int $signal): void
    {
        $this->shouldStop = true;
    }
}

interrupted 方法仅在工作进程于作业运行期间收到进程信号时才会被调用。它并非超时的替代方案，也并非作业的 failed 方法的替代方案。

作业过期与超时

作业过期

在 config/queue.php 配置文件中，每个队列连接都定义了一个 retry_after 选项。该选项指定队列连接在重试正在处理的作业之前应等待的秒数。例如，若将 retry_after 的值设为 90，则当作业在未被释放或删除的情况下处理时长达到 90 秒时，该作业将被重新释放回队列。通常，你应将 retry_after 的值设置为作业合理完成处理所需的最大秒数。

唯一不包含 retry_after 值的队列连接是 Amazon SQS。SQS 会依据 默认可见性超时 重试作业，该超时设置在 AWS 控制台中进行管理。

工作进程超时

queue:work Artisan 命令提供了一个 --timeout 选项。默认情况下，--timeout 的值为 60 秒。如果某个作业的处理时长超过了超时值所指定的秒数，处理该作业的工作进程会报错退出。通常，服务器上配置的进程管理器会自动重启该工作进程：

php artisan queue:work --timeout=60

retry_after 配置选项与 --timeout CLI 选项有所不同，但二者协同作用，以确保作业不会丢失，且每个作业仅被成功处理一次。

--timeout 的值应始终比 retry_after 配置值至少短几秒。这能确保处理卡死作业的工作进程总是在作业被重试之前终止。若 --timeout 选项的值大于 retry_after 配置值，你的作业可能会被处理两次。

暂停与恢复队列工作进程

有时，你可能需要暂时阻止队列工作进程处理新作业，但又不想完全停止该工作进程。例如，你可能希望在系统维护期间暂停作业处理。Laravel 提供了 queue:pause 和 queue:continue Artisan 命令，用于暂停和恢复队列工作进程。

若要暂停特定队列，请提供队列连接名称和队列名称：

php artisan queue:pause database:default

在此示例中，database 是队列连接名称，default 是队列名称。队列一旦暂停，所有从该队列处理作业的工作进程会继续完成当前作业，但在队列恢复之前不会再接收任何新作业。

若要暂停所有连接上的每个队列的作业处理，请使用 --all 选项：

php artisan queue:pause --all

若要恢复已暂停队列的作业处理，请使用 queue:continue 命令：

php artisan queue:continue database:default

若要恢复所有连接上的每个队列的作业处理，请使用带有 --all 选项的 queue:resume 命令：

php artisan queue:resume --all

队列恢复后，工作进程将立即开始处理该队列的新作业。恢复所有队列的操作不会恢复之前单独暂停的队列。请注意，暂停队列并不会停止工作进程本身——它只是阻止工作进程从指定队列处理新作业。

工作进程重启与暂停信号

默认情况下，队列工作进程在每次作业迭代时都会轮询缓存驱动，以检查重启和暂停信号。尽管这种轮询对于响应 queue:restart 和 queue:pause 命令至关重要，但它确实会带来一定的性能开销。

若你需要优化性能且不需要这些中断功能，可通过在 Queue 门面（Facade）上调用 withoutInterruptionPolling 方法，全局禁用该轮询。通常，此操作应在 AppServiceProvider 的 boot 方法中执行：

use Illuminate\Support\Facades\Queue;

/**
 * 引导任何应用服务。
 */
public function boot(): void
{
    Queue::withoutInterruptionPolling();
}

此外，你还可以通过设置 Illuminate\Queue\Worker 类上的静态属性 $restartable 或 $pausable，分别单独禁用重启或暂停轮询：

use Illuminate\Queue\Worker;

/**
 * 引导任何应用服务。
 */
public function boot(): void
{
    Worker::$restartable = false;
    Worker::$pausable = false;
}

当中断轮询被禁用后，工作进程将不会响应 queue:restart 或 queue:pause 命令（具体取决于被禁用的功能）。

Supervisor 配置

在生产环境中，你需要一种方式来确保 queue:work 进程持续运行。queue:work 进程可能会因多种原因停止运行，例如工作进程超时，或执行了 queue:restart 命令。

因此，你需要配置一个进程监视器，使其能够检测到 queue:work 进程何时退出，并自动重启这些进程。此外，进程监视器还允许你指定希望并发运行的 queue:work 进程数量。Supervisor 是一种常用于 Linux 环境的进程监视器，我们将在后续文档中讨论如何对其进行配置。

安装 Supervisor

Supervisor 是适用于 Linux 操作系统的进程监视器，当你的 queue:work 进程出现故障时，它会自动重启这些进程。要在 Ubuntu 上安装 Supervisor，你可以使用以下命令：

sudo apt-get install supervisor

如果自行配置和管理 Supervisor 让你感到困难，不妨考虑使用 Laravel Cloud，它为运行 Laravel 队列工作进程提供了一个完全托管的平台。

配置 Supervisor

Supervisor 的配置文件通常存放在 /etc/supervisor/conf.d 目录下。在此目录中，你可以创建任意数量的配置文件，以指示 Supervisor 如何监视你的进程。例如，我们可以创建一个 laravel-worker.conf 文件，用于启动并监视 queue:work 进程：

[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /home/forge/app.com/artisan queue:work --sleep=3 --tries=3 --max-time=3600
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=forge
numprocs=8
redirect_stderr=true
stdout_logfile=/home/forge/app.com/worker.log
stopwaitsecs=3600

在此示例中，numprocs 指令将指示 Supervisor 运行 8 个 queue:work 进程并对它们进行监视，若这些进程出现故障，则会自动重启。你应根据自身所需的队列连接和工作进程选项，修改配置中的 command 指令。

你应该确保 stopwaitsecs 的值大于你耗时最长的作业所消耗的秒数。否则，Supervisor 可能会在作业处理完成前将其终止。

启动 Supervisor

配置文件创建完成后，你可以使用以下命令更新 Supervisor 配置并启动这些进程：

sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start "laravel-worker:*"

若想进一步了解 Supervisor，请查阅 Supervisor 文档。

处理失败的作业

有时，排队任务可能会失败。别担心，事情不可能总是一帆风顺！Laravel 提供了一种便捷的方式来指定任务应尝试执行的最大次数。当异步任务超过此尝试次数后，它将被插入到 failed_jobs 数据库表中。同步派发的任务若失败，不会存储在此表中，其异常会立即由应用程序处理。

在新的 Laravel 应用中，通常已包含用于创建 failed_jobs 表的迁移文件。不过，如果你的应用中没有该表的迁移文件，可以使用 make:queue-failed-table 命令来创建迁移：

php artisan make:queue-failed-table

php artisan migrate

运行队列工作进程时，可以通过 queue:work 命令上的 --tries 参数指定任务应尝试执行的最大次数。如果未为 --tries 选项指定值，任务仅会被尝试一次，或者按照任务类的 Tries 属性指定的次数尝试：

php artisan queue:work redis --tries=3

使用 --backoff 选项，可以指定 Laravel 在重试遇到异常的任务前应等待的秒数。默认情况下，任务会立即重新释放到队列中，以便再次尝试：

php artisan queue:work redis --tries=3 --backoff=3

如果你想针对单个任务配置 Laravel 在重试遇到异常的任务前应等待的秒数，可以在任务类中使用 Backoff 属性：

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Backoff;

#[Backoff(3)]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

如果需要更复杂的逻辑来确定任务的退避时间，可以在任务类中定义一个 backoff 方法：

/**
 * 计算重试任务前要等待的秒数。
 */
public function backoff(): int
{
    return 3;
}

通过定义退避值数组，可以轻松配置“指数”退避。在此示例中，第一次重试的延迟时间为 1 秒，第二次重试为 5 秒，第三次重试为 10 秒，若仍有剩余尝试次数，后续每次重试的延迟时间均为 10 秒：

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Backoff;

#[Backoff([1, 5, 10])]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

失败任务的清理

当某个任务失败时，你可能希望向用户发送提醒，或撤销任务已部分完成的操作。为此，可以在任务类中定义一个 failed 方法。导致任务失败的 Throwable 实例将被传递给 failed 方法：

<?php

namespace App\Jobs;

use App\Models\Podcast;
use App\Services\AudioProcessor;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Throwable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public Podcast $podcast,
    ) {}

    /**
     * Execute the job.
     */
    public function handle(AudioProcessor $processor): void
    {
        // Process uploaded podcast...
    }

    /**
     * Handle a job failure.
     */
    public function failed(?Throwable $exception): void
    {
        // Send user notification of failure, etc...
    }
}

在调用 failed 方法之前，会实例化一个新的任务实例；因此，handle 方法中可能发生的任何类属性修改都会丢失。

失败的任务并不一定是遇到未处理异常的任务。当任务耗尽所有允许的尝试次数时，也可被视为失败。这些尝试次数可能以多种方式耗尽：

• 任务超时。
• 任务在执行过程中遇到未处理的异常。
• 任务被手动或由中间件释放回队列。

如果最后一次尝试因任务执行过程中抛出异常而失败，该异常将被传递给任务的 failed 方法。不过，如果任务失败是因为达到了允许的最大尝试次数，$exception 将是 Illuminate\Queue\MaxAttemptsExceededException 的一个实例。同样，如果任务因超出配置的超时时间而失败，$exception 将是 Illuminate\Queue\TimeoutExceededException 的一个实例。

重试失败的任务

若要查看所有已插入 failed_jobs 数据库表中的失败任务，可以使用 queue:failed Artisan 命令：

php artisan queue:failed

queue:failed 命令将列出任务 ID、连接、队列、失败时间以及任务的其他信息。任务 ID 可用于重试失败的任务。例如，要重试 ID 为 ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece 的失败任务，请执行以下命令：

php artisan queue:retry ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece

如有需要，可以向命令传递多个 ID：

php artisan queue:retry ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece 91401d2c-0784-4f43-824c-34f94a33c24d

也可以重试特定队列中的所有失败任务：

php artisan queue:retry --queue=name

若要重试所有失败任务，执行 queue:retry 命令并传入 all 作为 ID：

php artisan queue:retry all

如果想删除一个失败任务，可以使用 queue:forget 命令：

php artisan queue:forget 91401d2c-0784-4f43-824c-34f94a33c24d

当使用 Horizon 时，应使用 horizon:forget 命令删除失败任务，而非 queue:forget 命令。

若要从 failed_jobs 表中删除所有失败任务，可以使用 queue:flush 命令：

php artisan queue:flush

queue:flush 命令会从队列中移除所有失败任务记录，无论这些失败任务存在多久。可以使用 --hours 选项仅删除在特定小时数前或更早失败的任务：

php artisan queue:flush --hours=48

忽略缺失的模型

当将 Eloquent 模型注入任务时，模型会自动序列化，然后被放入队列，在任务处理时再从数据库中重新检索。不过，如果在任务等待工作进程处理期间模型已被删除，任务可能会因 ModelNotFoundException 而失败。

为方便起见，可以选择在任务类中使用 DeleteWhenMissingModels 属性，自动删除缺失模型的任务。当存在此属性时，Laravel 会悄悄丢弃该任务，不会抛出异常：

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\DeleteWhenMissingModels;

#[DeleteWhenMissingModels]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

清理失败任务

可以通过调用 queue:prune-failed Artisan 命令，清理应用 failed_jobs 表中的记录：

php artisan queue:prune-failed

默认情况下，所有超过 24 小时的失败任务记录都会被清理。如果向命令提供 --hours 选项，则只会保留在过去 N 小时内插入的失败任务记录。例如，以下命令将删除所有在 48 小时前插入的失败任务记录：

php artisan queue:prune-failed --hours=48

将失败任务存储在 DynamoDB 中

Laravel 还支持将失败任务记录存储在 DynamoDB 中，而非关系型数据库表中。不过，你必须手动创建一个 DynamoDB 表来存储所有失败任务记录。通常，该表应命名为 failed_jobs，但应根据应用 queue 配置文件中 queue.failed.table 配置项的值来命名该表。

failed_jobs 表应具有一个名为 application 的字符串主分区键和一个名为 uuid 的字符串主排序键。该键的 application 部分将包含应用程序的名称，此名称由应用程序 app 配置文件中的 name 配置值定义。由于应用程序名称是 DynamoDB 表键的一部分，因此您可以使用同一张表存储多个 Laravel 应用程序的失败任务。

此外，请确保安装 AWS SDK，以便您的 Laravel 应用程序能够与 Amazon DynamoDB 进行通信：

composer require aws/aws-sdk-php

接下来，将 queue.failed.driver 配置选项的值设置为 dynamodb。此外，您应在失败任务配置数组中定义 key、secret 和 region 配置选项。这些选项将用于进行 AWS 身份验证。使用 dynamodb 驱动时，queue.failed.database 配置选项是不必要的：

'failed' => [
    'driver' => env('QUEUE_FAILED_DRIVER', 'dynamodb'),
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'table' => 'failed_jobs',
],

禁用失败任务存储

您可以通过将 queue.failed.driver 配置选项的值设置为 null，指示 Laravel 丢弃失败任务而不进行存储。通常，这可以通过 QUEUE_FAILED_DRIVER 环境变量来实现：

QUEUE_FAILED_DRIVER=null

失败任务事件

如果您想要注册一个在任务失败时会被调用的事件监听器，可以使用 Queue 门面的 failing 方法。例如，我们可以在 Laravel 自带的 AppServiceProvider 的 boot 方法中为该事件绑定一个闭包：

<?php

namespace App\Providers;

use Illuminate\Support\Facades\Queue;
use Illuminate\Support\ServiceProvider;
use Illuminate\Queue\Events\JobFailed;

class AppServiceProvider extends ServiceProvider
{
    /**
     * Register any application services.
     */
    public function register(): void
    {
        // ...
    }

    /**
     * Bootstrap any application services.
     */
    public function boot(): void
    {
        Queue::failing(function (JobFailed $event) {
            // $event->connectionName
            // $event->job
            // $event->exception
        });
    }
}

清除队列中的任务

当使用 Horizon 时，应使用 horizon:clear 命令清除队列中的任务，而非 queue:clear 命令。

如果您想从默认连接的默认队列中删除所有任务，可以使用 queue:clear Artisan 命令来实现：

php artisan queue:clear

您还可以提供 connection 参数和 queue 选项，以从特定的连接和队列中删除任务：

php artisan queue:clear redis --queue=emails

清除队列中的任务功能仅适用于 SQS、Redis 和数据库队列驱动。此外，SQS 消息删除过程最多需要 60 秒，因此在您清除队列后 60 秒内发送到 SQS 队列的任务也可能会被删除。

监控队列

如果队列突然涌入大量任务，可能会导致队列负载过高，使任务的完成等待时间过长。若有需要，Laravel 可以在队列任务数量超过指定阈值时向您发出提醒。

首先，您应安排 queue:monitor 命令每分钟运行一次。该命令接受您想要监控的队列名称以及您期望的任务数量阈值：

php artisan queue:monitor redis:default,redis:deployments --max=100

仅安排此命令并不足以触发通知，以提醒您队列负载过高的状态。当该命令检测到某个队列的任务数量超过您设定的阈值时，会派发一个 Illuminate\Queue\Events\QueueBusy 事件。您可以在应用程序的 AppServiceProvider 中监听此事件，以便向您或您的开发团队发送通知：

use App\Notifications\QueueHasLongWaitTime;
use Illuminate\Queue\Events\QueueBusy;
use Illuminate\Support\Facades\Event;
use Illuminate\Support\Facades\Notification;

/**
 * Bootstrap any application services.
 */
public function boot(): void
{
    Event::listen(function (QueueBusy $event) {
        Notification::route('mail', '[email protected]')
            ->notify(new QueueHasLongWaitTime(
                $event->connectionName,
                $event->queue,
                $event->size
            ));
    });
}

测试

在测试派发任务的代码时，您可能希望指示 Laravel 不要实际执行任务本身，因为任务的代码可以直接且独立于派发它的代码进行测试。当然，要测试任务本身，您可以实例化一个任务对象，并在测试中直接调用 handle 方法。

您可以使用 Queue 门面的 fake 方法来防止排队的任务被真正推送到队列中。调用 Queue 门面的 fake 方法后，您就可以断言应用程序是否尝试将任务推送到队列：

<?php

use App\Jobs\AnotherJob;
use App\Jobs\ShipOrder;
use Illuminate\Support\Facades\Queue;

test('orders can be shipped', function () {
    Queue::fake();

    // Perform order shipping...

    // Assert that no jobs were pushed...
    Queue::assertNothingPushed();

    // Assert a job was pushed to a given queue...
    Queue::assertPushedOn('queue-name', ShipOrder::class);

    // Assert a job was pushed
    Queue::assertPushed(ShipOrder::class);

    // Assert a job was pushed exactly once...
    Queue::assertPushedOnce(ShipOrder::class);

    // Assert a job was pushed twice...
    Queue::assertPushedTimes(ShipOrder::class, 2);

    // Assert a job was not pushed...
    Queue::assertNotPushed(AnotherJob::class);

    // Assert that a closure was pushed to the queue...
    Queue::assertClosurePushed();

    // Assert that a closure was not pushed...
    Queue::assertClosureNotPushed();

    // Assert the total number of jobs that were pushed...
    Queue::assertCount(3);
});

<?php

namespace Tests\Feature;

use App\Jobs\AnotherJob;
use App\Jobs\ShipOrder;
use Illuminate\Support\Facades\Queue;
use Tests\TestCase;

class ExampleTest extends TestCase
{
    public function test_orders_can_be_shipped(): void
    {
        Queue::fake();

        // Perform order shipping...

        // Assert that no jobs were pushed...
        Queue::assertNothingPushed();

        // Assert a job was pushed to a given queue...
        Queue::assertPushedOn('queue-name', ShipOrder::class);

        // Assert a job was pushed
        Queue::assertPushed(ShipOrder::class);

        // Assert a job was pushed exactly once...
        Queue::assertPushedOnce(ShipOrder::class);

        // Assert a job was pushed twice...
        Queue::assertPushedTimes(ShipOrder::class, 2);

        // Assert a job was not pushed...
        Queue::assertNotPushed(AnotherJob::class);

        // Assert that a closure was pushed to the queue...
        Queue::assertClosurePushed();

        // Assert that a closure was not pushed...
        Queue::assertClosureNotPushed();

        // Assert the total number of jobs that were pushed...
        Queue::assertCount(3);
    }
}

您可以向 assertPushed、assertNotPushed、assertClosurePushed 或 assertClosureNotPushed 方法传入一个闭包，以便断言已推送的任务通过了给定的“真值测试”。如果有至少一个已推送的任务通过了给定的真值测试，则断言将成功：

use Illuminate\Queue\CallQueuedClosure;

Queue::assertPushed(function (ShipOrder $job) use ($order) {
    return $job->order->id === $order->id;
});

Queue::assertClosurePushed(function (CallQueuedClosure $job) {
    return $job->name === 'validate-order';
});

此外，您可以将一组预期的属性值作为第二个参数传递给 assertPushed 或 assertNotPushed，或者作为第三个参数传递给 assertPushedOn：

Queue::assertPushed(ShipOrder::class, ['order' => $order]);

Queue::assertPushedOn('shipping', ShipOrder::class, ['order' => $order]);

所有指定的属性都必须匹配。值将采用严格相等的方式进行比较，而 Eloquent 模型则使用其 is 方法进行比较。Bus 门面的 assertDispatched、assertNotDispatched、assertDispatchedSync 和 assertDispatchedAfterResponse 方法也将一组预期的属性值作为第二个参数接收。

伪造部分任务

如果仅需伪造特定的任务，同时让其他任务正常执行，可以将应被伪造的任务类名传递给 fake 方法：

test('orders can be shipped', function () {
    Queue::fake([
        ShipOrder::class,
    ]);

    // 执行订单发货……

    // 断言某任务被推送了两次……
    Queue::assertPushedTimes(ShipOrder::class, 2);
});

public function test_orders_can_be_shipped(): void
{
    Queue::fake([
        ShipOrder::class,
    ]);

    // 执行订单发货……

    // 断言某任务被推送了两次……
    Queue::assertPushedTimes(ShipOrder::class, 2);
}

你可以使用 except 方法伪造除指定任务之外的所有任务：

Queue::fake()->except([
    ShipOrder::class,
]);

测试任务链

要测试任务链，需要利用 Bus 门面的伪造功能。Bus 门面的 assertChained 方法可用于断言任务链已被分发。assertChained 方法将一个任务链数组作为其第一个参数：

use App\Jobs\RecordShipment;
use App\Jobs\ShipOrder;
use App\Jobs\UpdateInventory;
use Illuminate\Support\Facades\Bus;

Bus::fake();

// ...

Bus::assertChained([
    ShipOrder::class,
    RecordShipment::class,
    UpdateInventory::class
]);

正如上方示例所示，任务链数组可以是任务类名组成的数组。不过，你也可以提供一个包含实际任务实例的数组。这样做时，Laravel 会确保这些任务实例与应用所分发的链式任务属于相同的类，并且具有相同的属性值：

Bus::assertChained([
    new ShipOrder,
    new RecordShipment,
    new UpdateInventory,
]);

你可以使用 assertDispatchedWithoutChain 方法来断言某个任务在推送时未附带任务链：

Bus::assertDispatchedWithoutChain(ShipOrder::class);

测试任务链的修改

如果链式任务向现有任务链的头部或尾部添加了任务，你可以使用任务的 assertHasChain 方法，断言该任务剩余的任务链符合预期：

$job = new ProcessPodcast;

$job->handle();

$job->assertHasChain([
    new TranscribePodcast,
    new OptimizePodcast,
    new ReleasePodcast,
]);

可以使用 assertDoesntHaveChain 方法断言任务的剩余任务链为空：

$job->assertDoesntHaveChain();

测试链式批次

如果你的任务链中包含一个任务批次，你可以通过在任务链断言中插入 Bus::chainedBatch 定义，来断言该链式批次符合预期：

use App\Jobs\ShipOrder;
use App\Jobs\UpdateInventory;
use Illuminate\Bus\PendingBatch;
use Illuminate\Support\Facades\Bus;

Bus::assertChained([
    new ShipOrder,
    Bus::chainedBatch(function (PendingBatch $batch) {
        return $batch->jobs->count() === 3;
    }),
    new UpdateInventory,
]);

测试任务批次

Bus 门面的 assertBatched 方法可用于断言任务批次已被分发。传递给 assertBatched 方法的闭包会接收一个 Illuminate\Bus\PendingBatch 实例，可用于检查批次中的任务：

use Illuminate\Bus\PendingBatch;
use Illuminate\Support\Facades\Bus;

Bus::fake();

// ...

Bus::assertBatched(function (PendingBatch $batch) {
    return $batch->name == 'Import CSV' &&
           $batch->jobs->count() === 10;
});

可以在待处理批次上使用 hasJobs 方法，以验证该批次包含预期的任务。该方法接收一个包含任务实例、类名或闭包的数组：

Bus::assertBatched(function (PendingBatch $batch) {
    return $batch->hasJobs([
        new ProcessCsvRow(row: 1),
        new ProcessCsvRow(row: 2),
        new ProcessCsvRow(row: 3),
    ]);
});

当使用闭包时，该闭包会接收任务实例。预期的任务类型将从闭包的类型提示中推断得出：

Bus::assertBatched(function (PendingBatch $batch) {
    return $batch->hasJobs([
        fn (ProcessCsvRow $job) => $job->row === 1,
        fn (ProcessCsvRow $job) => $job->row === 2,
        fn (ProcessCsvRow $job) => $job->row === 3,
    ]);
});

你可以使用 assertBatchCount 方法，断言指定数量的批次已被分发：

Bus::assertBatchCount(3);

你可以使用 assertNothingBatched 来断言没有分发任何批次：

Bus::assertNothingBatched();

测试任务/批次交互

此外，偶尔需要测试单个任务与其底层批次之间的交互。例如，可能需要测试某个任务是否取消了其所属批次的后续处理。为此，需要通过 withFakeBatch 方法为任务分配一个伪造的批次。withFakeBatch 方法会返回一个元组，其中包含任务实例和伪造的批次：

[$job, $batch] = (new ShipOrder)->withFakeBatch();

$job->handle();

$this->assertTrue($batch->cancelled());
$this->assertEmpty($batch->added);

测试任务/队列交互

有时，你需要测试队列任务是否会将自身重新释放回队列，或者需要测试任务是否删除了自身。你可以通过实例化任务并调用 withFakeQueueInteractions 方法来测试这些队列交互。

一旦任务的队列交互被伪造，即可调用任务的 handle 方法。调用任务后，可使用多种断言方法来验证任务的队列交互：

use App\Exceptions\CorruptedAudioException;
use App\Jobs\ProcessPodcast;

$job = (new ProcessPodcast)->withFakeQueueInteractions();

$job->handle();

$job->assertReleased(delay: 30);
$job->assertDeleted();
$job->assertNotDeleted();
$job->assertFailed();
$job->assertFailedWith(CorruptedAudioException::class);
$job->assertNotFailed();

任务事件

通过 Queue 门面上的 before 和 after 方法，你可以指定在队列任务处理前后执行的回调。这些回调非常适合用于执行额外的日志记录，或者为仪表盘更新统计数据。通常，应当在 服务提供者的 boot 方法中调用这些方法。例如，可以使用 Laravel 自带的 AppServiceProvider：

<?php

namespace App\Providers;

use Illuminate\Support\Facades\Queue;
use Illuminate\Support\ServiceProvider;
use Illuminate\Queue\Events\JobProcessed;
use Illuminate\Queue\Events\JobProcessing;

class AppServiceProvider extends ServiceProvider
{
    /**
     * 注册应用服务。
     */
    public function register(): void
    {
        // ...
    }

    /**
     * 启动应用服务。
     */
    public function boot(): void
    {
        Queue::before(function (JobProcessing $event) {
            // $event->connectionName
            // $event->job
            // $event->job->payload()
        });

        Queue::after(function (JobProcessed $event) {
            // $event->connectionName
            // $event->job
            // $event->job->payload()
        });
    }
}

通过 Queue 门面上的 looping 方法，你可以指定在工作进程尝试从队列获取任务之前执行的回调。例如，你可以注册一个闭包，用于回滚先前失败任务遗留的未关闭事务：

use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Queue;

Queue::looping(function () {
    while (DB::transactionLevel() > 0) {
        DB::rollBack();
    }
});

当队列工作进程无法从队列中获取任务时，Laravel 还会分发 Illuminate\Queue\Events\WorkerIdle 事件：

use Illuminate\Queue\Events\WorkerIdle;
use Illuminate\Support\Facades\Event;

Event::listen(function (WorkerIdle $event) {
    // $event->connectionName
    // $event->queue
    // $event->workerOptions
});