バージョン: 13.x

Queues

Introduction
- Connections vs. Queues
- Driver Notes and Prerequisites
Creating Jobs
Job Middleware
Dispatching Jobs
Job Batching
Queueing Closures
Running the Queue Worker
Supervisor Configuration
Dealing With Failed Jobs
Clearing Jobs From Queues
Monitoring Your Queues
Testing
Job Events

Introduction

Web アプリケーションの構築中に、アップロードされた CSV ファイルの解析や保存など、通常の Web リクエストでは実行するには時間がかかりすぎるタスクがいくつか発生する場合があります。ありがたいことに、Laravel ではバックグラウンドで処理できるキューに入れられたジョブを簡単に作成できます。時間のかかるタスクをキューに移動することで、アプリケーションは Web リクエストに驚異的な速度で応答し、顧客により良いユーザーエクスペリエンスを提供できるようになります。

Laravel キューは、Amazon SQS、Redis、さらにはリレーショナルデータベースなど、さまざまな異なるキューバックエンドにわたって統合されたキュー API を提供します。

Laravel のキュー構成オプションは、アプリケーションの config/queue.php 構成ファイルに保存されます。このファイルには、データベース、Amazon SQS、Redis、Beanstalkd ドライバや、(開発またはテスト中に使用する) ジョブを即時に実行する同期ドライバなど、フレームワークに含まれる各キュードライバの接続構成が含まれています。キューに入れられたジョブを破棄する null キュードライバも含まれています。

Laravel Horizon は、Redis で動作するキュー向けの美しいダッシュボードおよび設定システムです。詳しくは、Horizon documentation を参照してください。

Connections vs. Queues

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

アプリケーションによっては、ジョブを複数のキューにプッシュする必要がなく、代わりに 1 つの単純なキューを持つことを好む場合があります。ただし、Laravel キューワーカーでは優先度によってどのキューを処理するかを指定できるため、ジョブを複数のキューにプッシュすることは、ジョブの処理方法に優先順位を付けたり、分割したりしたいアプリケーションにとって特に便利です。たとえば、ジョブを high キューにプッシュする場合、より高い処理優先度を与えるワーカーを実行できます。

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

Driver Notes and Prerequisites

Database

database キュードライバを使用するには、ジョブを保持するためのデータベーステーブルが必要です。通常、このテーブルは Laravel のデフォルトに含まれる 0001_01_01_000002_create_jobs_table.php database migration によって作成されます。ただし、アプリケーションにこのマイグレーションが含まれていない場合は、make:queue-table Artisan コマンドを使用して作成できます。

php artisan make:queue-table

php artisan migrate

Redis

redis キュードライバを使用するには、config/database.php 構成ファイルで Redis データベース接続を構成する必要があります。

redis キュードライバでは、serializer および compression Redis オプションはサポートされていません。

Redis Cluster

Redis キュー接続で Redis Cluster を使用する場合、キュー名には key hash tag が含まれている必要があります。これは、特定のキューのすべての 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,
],

Blocking

Redis キューを使用する場合、block_for 構成オプションを使用して、ドライバがワーカーループを繰り返して Redis データベースを再ポーリングする前に、ジョブが使用可能になるまで待機する時間を指定できます。

キューの負荷に基づいてこの値を調整すると、新しいジョブを求めて Redis データベースを継続的にポーリングするよりも効率的になる場合があります。たとえば、値を 5 に設定して、ジョブが使用可能になるまでドライバが 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 Overflow Storage

Amazon SQS は、キューに入れられたメッセージペイロードの最大サイズを制限します。この制限を超える可能性のあるペイロードを含むジョブをディスパッチする必要がある場合は、特大の SQS ペイロードをキャッシュストアに保存し、代わりに SQS 経由でポインタを送信するように Laravel を設定できます。この機能を有効にするには、overflow 配列を SQS キュー接続構成に追加します。

'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 オーバーフローストレージを構成する必要があります。

Other Driver Prerequisites

リストされているキュードライバには次の依存関係が必要です。これらの依存関係は、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

Creating Jobs

Generating Job Classes

デフォルトでは、アプリケーションのキューに入れられるジョブはすべて、app/Jobs ディレクトリに保存されます。 app/Jobs ディレクトリが存在しない場合は、make:job Artisan コマンドを実行すると作成されます。

php artisan make:job ProcessPodcast

生成されたクラスは Illuminate\Contracts\Queue\ShouldQueue インターフェイスを実装し、非同期で実行するにはジョブをキューにプッシュする必要があることを Laravel に示します。

Job stubs は stub publishing を使用してカスタマイズできます。

Class Structure

ジョブクラスは非常に単純で、通常はジョブがキューによって処理されるときに呼び出される 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;

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

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

この例では、キューに入れるジョブのコンストラクタに Eloquent model を直接渡せることに注目してください。ジョブが使用している Queueable トレイトにより、ジョブの処理時に Eloquent モデルとロード済みのリレーションが適切にシリアライズおよびデシリアライズされます。

キューに入れられたジョブがコンストラクターで Eloquent モデルを受け入れる場合、モデルの識別子のみがキューにシリアル化されます。ジョブが実際に処理されると、キューシステムは完全なモデルインスタンスとそのロードされた関係をデータベースから自動的に再取得します。モデルのシリアル化に対するこのアプローチにより、はるかに小さいジョブペイロードをキュードライバに送信できるようになります。

`handle` Method Dependency Injection

ジョブがキューによって処理されると、handle メソッドが呼び出されます。ジョブの handle メソッドでは、依存関係をタイプヒントで指定できます。Laravel の service container が、これらの依存関係を自動的に注入します。

コンテナが handle メソッドに依存関係を注入する方法を完全に制御したい場合は、コンテナの bindMethod メソッドを使用できます。bindMethod メソッドは、ジョブとコンテナを受け取るコールバックを指定します。コールバック内では、自由な方法で handle メソッドを呼び出せます。通常、このメソッドは App\Providers\AppServiceProvider の service provider の 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 として正しくシリアライズできない場合があります。

Queued Relationships

ジョブがキューに入れられると、ロードされたすべての Eloquent モデルの関係もシリアル化されるため、シリアル化されたジョブ文字列が非常に大きくなる場合があります。さらに、ジョブが逆シリアル化され、モデルの関係がデータベースから再取得されると、それらは完全に取得されます。ジョブキューイングプロセス中にモデルがシリアル化される前に適用された以前の関係制約は、ジョブが逆シリアル化されると適用されません。したがって、特定の関係のサブセットを操作したい場合は、キューに入れられたジョブ内でその関係を再制約する必要があります。

または、リレーションがシリアル化されないようにするには、プロパティ値を設定するときにモデルで withoutRelations メソッドを呼び出します。このメソッドは、ロードされた関係を持たないモデルのインスタンスを返します。

/**
 * Create a new job instance.
 */
public function __construct(
    Podcast $podcast,
) {
    $this->podcast = $podcast->withoutRelations();
}

残りの関係を保持したまま特定の関係のみを削除する必要がある場合は、withoutRelation メソッドを使用できます。

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

PHP constructor property promotion を使用していて、Eloquent モデルの関係をシリアル化しないことを示したい場合は、WithoutRelations 属性を使用できます。

use Illuminate\Queue\Attributes\WithoutRelations;

/**
 * Create a new job instance.
 */
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;

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

ジョブが単一のモデルではなく Eloquent モデルのコレクションまたは配列を受け取った場合、ジョブが逆シリアル化されて実行されたときに、そのコレクション内のモデルの関係は復元されません。これは、多数のモデルを処理するジョブでの過剰なリソースの使用を防ぐためです。

Unique Jobs

ユニークジョブには、locksをサポートするキャッシュドライバが必要です。現在、memcached、redis、dynamodb、database、file、array のキャッシュドライバがアトミックロックをサポートしています。

バッチ内のジョブには、一意なジョブ制約は適用されません。

場合によっては、特定のジョブのインスタンスが常に 1 つだけキューに存在するようにしたい場合があります。これを行うには、ジョブクラスに 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
{
    /**
     * The product instance.
     *
     * @var \App\Models\Product
     */
    public $product;

    /**
     * Get the unique ID for the job.
     */
    public function uniqueId(): string
    {
        return $this->product->id;
    }
}

上の例では、UpdateSearchIndex ジョブは製品 ID によって一意です。したがって、同じ製品 ID を持つジョブの新しいディスパッチは、既存のジョブの処理が完了するまで無視されます。さらに、既存のジョブが 1 時間以内に処理されない場合、一意のロックが解除され、同じ一意のキーを持つ別のジョブをキューにディスパッチできるようになります。

アプリケーションが複数のWebサーバーやコンテナからジョブをディスパッチする場合は、Laravelがジョブの一意性を正確に判定できるよう、すべてのサーバーが同じ中央キャッシュサーバーと通信することを確認してください。

Keeping Jobs Unique Until Processing Begins

デフォルトでは、ジョブが処理を完了するか、すべての再試行に失敗すると、固有のジョブは「ロック解除」されます。ただし、ジョブが処理される直前にロックを解除したい場合もあります。これを実現するには、ジョブで ShouldBeUnique コントラクトの代わりに ShouldBeUniqueUntilProcessing コントラクトを実装する必要があります。

<?php

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

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

Unique Job Locks

内部では、ShouldBeUnique ジョブがディスパッチされると、Laravel は uniqueId キーを使って lock の取得を試みます。すでにロックが取得されている場合、ジョブはディスパッチされません。このロックは、ジョブの処理が完了したとき、またはすべての再試行に失敗したときに解放されます。デフォルトでは、Laravel はこのロックの取得にデフォルトのキャッシュドライバを使用します。ただし、ロックの取得に別のドライバを使いたい場合は、使用するキャッシュドライバを返す uniqueVia メソッドを定義できます。

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

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

    /**
     * Get the cache driver for the unique job lock.
     */
    public function uniqueVia(): Repository
    {
        return Cache::driver('redis');
    }
}

ジョブの同時処理数を制限するだけでよい場合は、WithoutOverlapping ジョブミドルウェアを使用してください。

Debounced Jobs

場合によっては、同じジョブが短いウィンドウ内で何度もディスパッチされるときに、最新のディスパッチのみが実際に実行されるようにしたい場合があります。これを行うには、ジョブに 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;

    /**
     * Create a new job instance.
     */
    public function __construct(public int $productId)
    {
    }

    /**
     * Get the debounce ID for the job.
     */
    public function debounceId(): string
    {
        return (string) $this->productId;
    }
}

上記の例では、同じ製品に対して 30 秒以内に UpdateSearchIndex を繰り返しディスパッチするとジョブがデバウンスされ、最新のディスパッチのみが実行されるようになります。

頻繁に再ディスパッチされるジョブを延期できる期間を制限したい場合は、maxWait 引数を DebounceFor 属性に指定できます。

#[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 サーバーやコンテナからデバウンスされたジョブをディスパッチする場合は、すべてのサーバーが同じ中央キャッシュサーバーと通信していることを確認してください。

Encrypted Jobs

Laravelでは、encryptionによってジョブデータのプライバシーと完全性を確保できます。まず、ジョブクラスに ShouldBeEncrypted インターフェイスを追加してください。クラスにこのインターフェイスを追加すると、Laravelはジョブをキューへ投入する前に自動的に暗号化します。

<?php

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

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

Job Middleware

ジョブミドルウェアを使用すると、キューに入れられたジョブの実行にカスタムロジックをラップして、ジョブ自体の定型文を減らすことができます。たとえば、Laravel の Redis レート制限機能を活用して、5 秒ごとに 1 つのジョブのみの処理を許可する次の handle メソッドを考えてみましょう。

use Illuminate\Support\Facades\Redis;

/**
 * Execute the job.
 */
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
{
    /**
     * Process the queued job.
     *
     * @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);
            });
    }
}

見てのとおり、route middlewareと同様に、ジョブミドルウェアは処理中のジョブと、ジョブの処理を続けるために呼び出すコールバックを受け取ります。

make:job-middleware Artisan コマンドを使用して、新しいジョブミドルウェアクラスを生成できます。ジョブミドルウェアを作成した後、ジョブの middleware メソッドから返すことによって、ジョブミドルウェアをジョブにアタッチできます。このメソッドは、make:job Artisan コマンドによってスキャフォールディングされたジョブには存在しないため、ジョブクラスに手動で追加する必要があります。

use App\Jobs\Middleware\RateLimited;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited];
}

ジョブミドルウェアは、queueable event listeners、mailables、notificationsにも割り当てられます。

Rate Limiting

先ほど独自のジョブのレート制限ミドルウェアを作成する方法を説明しましたが、Laravel にはジョブのレートを制限するために利用できるレート制限ミドルウェアが組み込まれています。route rate limitersと同様に、ジョブのレートリミッタは RateLimiter ファサードの for メソッドを使用して定義します。

たとえば、プレミアム顧客にはそのような制限を課さず、ユーザーが 1 時間に 1 回データをバックアップできるようにしたい場合があります。これを実現するには、AppServiceProvider の boot メソッドで RateLimiter を定義できます。

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

/**
 * Bootstrap any application services.
 */
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;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited('backups')];
}

レート制限されたジョブをキューに解放しても、ジョブの総試行回数 attempts は増加します。それに応じてジョブクラスの Tries 属性と MaxExceptions 属性を調整することもできます。または、retryUntil method を使用して、ジョブが試行されなくなるまでの時間を定義することもできます。

releaseAfter メソッドを使用すると、キューに戻されたジョブが再試行されるまでに経過する必要がある秒数を指定することもできます。

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->releaseAfter(60)];
}

レートが制限されているときにジョブを再試行したくない場合は、dontRelease メソッドを使用できます。

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->dontRelease()];
}

Rate Limiting With 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')];

Preventing Job Overlaps

Laravel には、任意のキーに基づいてジョブの重複を防止できる Illuminate\Queue\Middleware\WithoutOverlapping ミドルウェアが含まれています。これは、キューに入れられたジョブが、一度に 1 つのジョブのみによって変更されるべきリソースを変更する場合に役立ちます。

たとえば、ユーザーのクレジットスコアを更新するキューに入れられたジョブがあり、同じユーザー ID に対してクレジットスコア更新ジョブが重複しないようにしたいとします。これを実現するには、ジョブの middleware メソッドから WithoutOverlapping ミドルウェアを返すことができます。

use Illuminate\Queue\Middleware\WithoutOverlapping;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new WithoutOverlapping($this->user->id)];
}

重複するジョブをキューに解放しても、ジョブの合計試行回数は増加します。それに応じて、ジョブクラスの Tries 属性と MaxExceptions 属性を調整することもできます。たとえば、Tries をデフォルトのまま 1 のままにすると、重複するジョブが後で再試行されなくなります。

同じタイプの重複するジョブはキューに戻されます。解放されたジョブが再試行されるまでに経過する必要がある秒数を指定することもできます。

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->releaseAfter(60)];
}

再試行されないように重複するジョブをすぐに削除したい場合は、dontRelease メソッドを使用できます。

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->dontRelease()];
}

WithoutOverlapping ミドルウェアは、Laravel のアトミックロック機能を利用しています。場合によっては、ロックが解放されずにジョブが予期せず失敗したり、タイムアウトになったりすることがあります。したがって、expireAfter メソッドを使用して、ロックの有効期限を明示的に定義できます。たとえば、以下の例は、ジョブの処理が開始されてから 3 分後に WithoutOverlapping ロックを解放するように Laravel に指示します。

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->expireAfter(180)];
}

WithoutOverlapping ミドルウェアには、locksをサポートするキャッシュドライバが必要です。現在、memcached、redis、dynamodb、database、file、array のキャッシュドライバがアトミックロックをサポートしています。

デフォルトでは、WithoutOverlapping ミドルウェアは、同じクラスの重複ジョブのみを防止します。したがって、2 つの異なるジョブクラスが同じロックキーを使用することはできますが、それらの重複は防止されません。ただし、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(),
        ];
    }
}

Throttling Exceptions

Laravel には、例外を抑制できる Illuminate\Queue\Middleware\ThrottlesExceptions ミドルウェアが含まれています。ジョブが指定された数の例外をスローすると、それ以降のジョブの実行試行は、指定された時間が経過するまで遅延されます。このミドルウェアは、不安定なサードパーティサービスと対話するジョブに特に役立ちます。

たとえば、例外をスローし始めるサードパーティ API と対話する、キューに入れられたジョブを想像してみましょう。例外を抑制するには、ジョブの middleware メソッドから ThrottlesExceptions ミドルウェアを返すことができます。通常、このミドルウェアは、time based attempts を実装するジョブと組み合わせる必要があります。

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

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new ThrottlesExceptions(10, 5 * 60)];
}

/**
 * Determine the time at which the job should timeout.
 */
public function retryUntil(): DateTime
{
    return now()->plus(minutes: 30);
}

ミドルウェアによって受け入れられる最初のコンストラクター引数は、ジョブの実行が制限される前にスローできる例外の数です。一方、2 番目のコンストラクター引数は、ジョブの実行が制限された後にジョブが再試行されるまでに経過する必要がある秒数です。上記のコード例では、ジョブが 10 回連続して例外をスローした場合、30 分の時間制限による制約を受けて、5 分間待ってからジョブを再試行します。

ジョブが例外をスローしたが、まだ例外しきい値に達していない場合、ジョブは通常、すぐに再試行されます。ただし、ミドルウェアをジョブにアタッチするときに backoff メソッドを呼び出すことで、そのようなジョブを遅延させる分数を指定できます。

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @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;

/**
 * Get the middleware the job should pass through.
 *
 * @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;

/**
 * Get the middleware the job should pass through.
 *
 * @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;

/**
 * Get the middleware the job should pass through.
 *
 * @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;

/**
 * Get the middleware the job should pass through.
 *
 * @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;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->report(
        fn (Throwable $throwable) => $throwable instanceof HttpClientException
    )];
}

Throttling Exceptions With 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')];

Releasing Jobs

Release ミドルウェアを使うと、ジョブを実行せずにキューへ戻せます。Release::when メソッドは、与えられた条件が true と評価された場合にジョブをキューへ戻し、Release::unless メソッドは、条件が false と評価された場合にジョブをキューへ戻します:

use Illuminate\Queue\Middleware\Release;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Release::when($condition, releaseAfter: 60),
    ];
}

ジョブをキューへ戻しても、そのジョブの総試行回数は増えます。ジョブクラスの Tries と MaxExceptions 属性は、それに応じて調整するとよいでしょう。

より複雑な条件評価を行う場合は、when および unless メソッドに Closure を渡すこともできます。

use Illuminate\Queue\Middleware\Release;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Release::when(function (): bool {
            return ! $this->order->isPaid();
        }, releaseAfter: 60),
    ];
}

Skipping Jobs

Skip ミドルウェアを使用すると、ジョブのロジックを変更することなく、ジョブをスキップ/削除するように指定できます。 Skip::when メソッドは、指定された条件が true と評価される場合にジョブを削除します。一方、Skip::unless メソッドは、条件が false と評価される場合にジョブを削除します。

use Illuminate\Queue\Middleware\Skip;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Skip::when($condition),
    ];
}

より複雑な条件評価を行う場合は、when メソッドと unless メソッドに Closure を渡すこともできます。

use Illuminate\Queue\Middleware\Skip;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Skip::when(function (): bool {
            return $this->shouldSkip();
        }),
    ];
}

Dispatching Jobs

ジョブクラスを作成したら、ジョブ自体で 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
{
    /**
     * Store a new podcast.
     */
    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 環境変数を変更することで、別のデフォルトのキュー接続を指定できます。

Delayed Dispatching

ジョブをキューワーカーによる処理にすぐに使用できないように指定したい場合は、ジョブをディスパッチするときに 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
{
    /**
     * Store a new podcast.
     */
    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 分です。

Synchronous Dispatching

ジョブをすぐに (同期的に) ディスパッチしたい場合は、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
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Create podcast...

        ProcessPodcast::dispatchSync($podcast);

        return redirect('/podcasts');
    }
}

Deferred Dispatching

遅延同期ディスパッチを使用すると、現在のプロセス中に、HTTP 応答がユーザーに送信された後で、処理されるジョブをディスパッチできます。これにより、ユーザーのアプリケーションエクスペリエンスを低下させることなく、「キューに入れられた」ジョブを同期的に処理できるようになります。同期ジョブの実行を延期するには、ジョブを deferred 接続にディスパッチします。

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

deferred 接続は、デフォルトの failover queue としても機能します。

同様に、background 接続は、HTTP 応答がユーザーに送信された後にジョブを処理します。ただし、ジョブは個別に生成された PHP プロセスで処理されるため、PHP-FPM / アプリケーションワーカーが別の受信 HTTP リクエストを処理できるようになります。

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

Bulk Dispatching

batch の追跡やコールバックを必要とせず、独立した多数のジョブを一度にディスパッチしたい場合は、Bus ファサードの bulk メソッドを使用できます。Laravel は、設定されたキュー接続とキュー名でジョブをグループ化し、各グループを適切なキューへ一括でプッシュします。

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

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

Preparing Jobs Before Dispatch

ジョブがキューにプッシュされる前にその状態を準備または検査する必要がある場合、ジョブは 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;

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

    /**
     * Prepare the job before dispatching.
     */
    public function prepareForDispatch(): bool
    {
        return collect($this->podcastIds)
            ->reject(fn (int $id) => Cache::has("podcast-syncing:{$id}"))
            ->isNotEmpty();
    }
}

Jobs & Database Transactions

データベーストランザクション内でジョブをディスパッチすることはまったく問題ありませんが、ジョブが実際に正常に実行できることを確認するために特別な注意を払う必要があります。トランザクション内でジョブをディスパッチする場合、親トランザクションがコミットされる前にジョブがワーカーによって処理される可能性があります。この問題が発生すると、データベーストランザクション中にモデルまたはデータベースレコードに加えた更新がまだデータベースに反映されていない可能性があります。さらに、トランザクション内で作成されたモデルやデータベースレコードはデータベースに存在しない可能性があります。

ありがたいことに、Laravel はこの問題を回避する方法をいくつか提供しています。まず、キュー接続の構成配列で after_commit 接続オプションを設定します。

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

after_commit オプションが true の場合、データベーストランザクション内でジョブをディスパッチできます。ただし、Laravel は、実際にジョブをディスパッチする前に、開いている親データベースのトランザクションがコミットされるまで待機します。もちろん、現在開いているデータベーストランザクションがない場合、ジョブはすぐにディスパッチされます。

トランザクション中に発生した例外によりトランザクションがロールバックされた場合、そのトランザクション中にディスパッチされたジョブは破棄されます。

after_commit 設定オプションを true にすると、キューに入れられたイベントリスナ、Mailable、通知、ブロードキャストイベントも、開いているすべてのデータベーストランザクションのコミット後にディスパッチされます。

Specifying Commit Dispatch Behavior Inline

after_commit キュー接続構成オプションを true に設定しない場合でも、開いているすべてのデータベーストランザクションがコミットされた後に特定のジョブをディスパッチする必要があることを示すことができます。これを実現するには、afterCommit メソッドをディスパッチ操作に連鎖させます。

use App\Jobs\ProcessPodcast;

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

同様に、after_commit 構成オプションが true に設定されている場合は、開いているデータベーストランザクションがコミットされるのを待たずに、特定のジョブを直ちにディスパッチする必要があることを指定できます。

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

Job Chaining

ジョブチェーンを使用すると、プライマリジョブが正常に実行された後に順番に実行する必要がある、キューに入れられたジョブのリストを指定できます。シーケンス内の 1 つのジョブが失敗すると、残りのジョブは実行されません。キューに入れられたジョブチェーンを実行するには、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() メソッドを使ってジョブを削除しても、チェーンされたジョブの処理は阻止されません。チェーンの実行が停止するのは、チェーン内のジョブが失敗した場合だけです。

Chain Connection and Queue

連鎖ジョブに使用する接続とキューを指定したい場合は、onConnection メソッドと onQueue メソッドを使用できます。これらのメソッドは、キューに入れられたジョブに別の接続/キューが明示的に割り当てられていない限り、使用する必要があるキュー接続とキュー名を指定します。

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

Adding Jobs to the Chain

場合によっては、既存のジョブチェーンに、そのチェーン内の別のジョブ内からジョブを先頭または末尾に追加する必要が生じることがあります。これは、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);
}

Chain Failures

ジョブをチェーンする場合、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 変数を使用しないでください。

Customizing the Queue and Connection

Dispatching to a Particular Queue

ジョブを異なるキューにプッシュすることで、キューに入れられたジョブを「分類」し、さまざまなキューに割り当てるワーカーの数に優先順位を付けることもできます。これは、キュー構成ファイルで定義されている別のキュー「接続」にジョブをプッシュするのではなく、単一の接続内の特定のキューにのみジョブをプッシュすることに注意してください。キューを指定するには、ジョブをディスパッチするときに 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');
    }
}

Dispatching to a Particular Connection

アプリケーションが複数のキュー接続と対話する場合は、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 Routing

Queue ファサードの route メソッドを使用して、特定のジョブクラスのデフォルトの接続とキューを定義できます。これは、ジョブで接続やキューを指定する必要がなく、特定のジョブが常に特定のキューを使用するようにしたい場合に便利です。

特定のジョブクラスをルーティングすることに加えて、インターフェイス、トレイト、または親クラスを route メソッドに渡すこともできます。これを行うと、インターフェイスを実装するジョブ、トレイトを使用するジョブ、または親クラスを拡張するジョブは、設定された接続とキューを自動的に使用します。

通常、サービスプロバイダの 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'], // Connection and queue
    ProcessVideo::class => 'videos', // Queue only (uses default connection)
]);

ジョブごとに、ジョブ側でキューのルーティングを上書きすることもできます。

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

ジョブに明示的な接続が設定されている場合は、転送先の接続よりも優先されます。

Specifying Max Job Attempts / Timeout Values

Max Attempts

ジョブ試行は、Laravel のキューシステムの中核概念であり、多くの高度な機能を強化します。最初は混乱するように思えるかもしれませんが、デフォルトの構成を変更する前に、それらがどのように機能するかを理解することが重要です。

ジョブがディスパッチされると、ジョブはキューにプッシュされます。次に、ワーカーがそれを拾い上げ、実行しようとします。これがジョブの1回の試行です。

ただし、試行は必ずしもジョブの handle メソッドが実行されたことを意味するわけではありません。試行はいくつかの方法で「消費」することもできます。

ジョブの実行中に未処理の例外が発生した場合。
$this->release() を使って、ジョブを手動でキューに戻した場合。
WithoutOverlapping や RateLimited などのミドルウェアがロックの取得に失敗し、ジョブをキューに戻した場合。
ジョブがタイムアウトした場合。
ジョブの handle メソッドが実行され、例外を投げずに完了した場合。

おそらく、ジョブの試行を無期限に続けることは望まないでしょう。したがって、Laravel では、ジョブを試行する回数や期間を指定するさまざまな方法が提供されています。

デフォルトでは、Laravel はジョブの実行を1回だけ試みます。ジョブで WithoutOverlapping や RateLimited などのミドルウェアを使用している場合、またはジョブを手動でキューに戻している場合は、tries オプションで許可する試行回数を増やす必要があるでしょう。

ジョブの最大試行回数を指定する方法の 1 つは、Artisan コマンドラインの --tries スイッチを使用することです。これは、処理中のジョブで試行回数が指定されていない限り、ワーカーによって処理されるすべてのジョブに適用されます。

php artisan queue:work --tries=3

ジョブが最大試行回数を超えた場合、そのジョブは「失敗した」ジョブとみなされます。失敗したジョブの処理の詳細については、failed job documentation を参照してください。 --tries=0 が queue:work コマンドに指定された場合、ジョブは無期限に再試行されます。

Tries 属性を使用して、ジョブクラス自体でジョブを試行できる最大回数を定義することで、より詳細なアプローチを採用することもできます。ジョブで最大試行回数が指定されている場合は、コマンドラインで指定した --tries 値よりも優先されます。

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Tries;

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

特定のジョブの最大試行回数を動的に制御する必要がある場合は、ジョブに tries メソッドを定義できます。

/**
 * Determine number of times the job may be attempted.
 */
public function tries(): int
{
    return 5;
}

Time Based Attempts

ジョブが失敗するまでに何回試行できるかを定義する代わりに、ジョブを試行しなくなる時間を定義することもできます。これにより、指定された時間枠内でジョブを何度でも試行できます。ジョブを試行しなくなる時刻を定義するには、retryUntil メソッドをジョブクラスに追加します。このメソッドは DateTime インスタンスを返す必要があります。

use DateTime;

/**
 * Determine the time at which the job should timeout.
 */
public function retryUntil(): DateTime
{
    return now()->plus(minutes: 10);
}

retryUntil と tries の両方が定義されている場合、Laravel は retryUntil メソッドを優先します。

Tries 属性や retryUntil メソッドは、queued event listeners や queued notifications にも定義できます。

Max Exceptions

場合によっては、ジョブを何度も試行できるが、(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;

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        Redis::throttle('key')->allow(10)->every(60)->then(function () {
            // Lock obtained, process the podcast...
        }, function () {
            // Unable to obtain lock...
            return $this->release(10);
        });
    }
}

この例では、アプリケーションが Redis ロックを取得できない場合、ジョブは 10 秒後に再試行可能になるようキューへ戻され、最大 25 回まで再試行され続けます。ただし、ジョブによって 3 つの未処理の例外がスローされた場合、ジョブは失敗します。

デフォルトでは、メモリ不足などによってワーカープロセスがクラッシュまたは強制終了されて試行が終了しても、ジョブの最大例外数にはカウントされません。このような試行も例外としてカウントするには、ジョブクラスに 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
{
    // ...
}

この属性が設定されている場合、ワーカーはジョブの処理中にアプリケーションのキャッシュへマーカーを保存します。次にジョブが試行された時点でもマーカーが残っていれば、前回の試行を例外としてカウントします。

Stopping Retries by Exception

場合によっては、例外がキュー投入されたジョブを再試行のために解放するのではなく、即座に失敗させるべきことを示すことがあります。アプリケーションの 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';
    });
})

Timeout

多くの場合、キューに入れられたジョブにかかる時間がおおよそわかっています。このため、Laravel では「タイムアウト」値を指定できます。デフォルトでは、タイムアウト値は 60 秒です。ジョブの処理がタイムアウト値で指定された秒数を超えた場合、ジョブを処理しているワーカーはエラーで終了します。通常、ワーカーは process manager configured on your server によって自動的に再起動されます。

ジョブを実行できる最大秒数は、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 接続などの IO ブロックプロセスが、指定されたタイムアウトを尊重しないことがあります。したがって、これらの機能を使用するときは、常にその API も使用してタイムアウトを指定するようにしてください。たとえば、Guzzle を使用する場合は、常に接続とリクエストのタイムアウト値を指定する必要があります。

ジョブのタイムアウトを指定するには、PCNTL PHP 拡張機能をインストールする必要があります。また、ジョブの timeout 値は必ず "retry after" 値より小さくしてください。そうしないと、ジョブの実行が実際に完了するかタイムアウトする前に、再試行される可能性があります。queue:work コマンドを --once オプション付きで呼び出した場合、--timeout オプションは効果がありません。

Failing on Timeout

タイムアウト時にジョブを failed としてマークする必要があることを示したい場合は、ジョブクラスで FailOnTimeout 属性を使用できます。

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\FailOnTimeout;

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

デフォルトでは、ジョブがタイムアウトすると1回分の試行を消費し、再試行が許可されている場合はキューに戻されます。ただし、タイムアウト時にジョブを失敗させるよう設定した場合、tries に設定した値にかかわらず再試行されません。

SQS FIFO and Fair Queues

Laravel は、Amazon SQS FIFO (First-In-First-Out) キューと fair キューをサポートしています。 FIFO キューを使用すると、メッセージの重複排除によって 1 回限りの処理を保証しながら、送信された順序どおりにジョブを処理できます。

FIFO キューには、どのジョブを並列処理できるかを決定するためのメッセージグループ ID が必要です。同じグループ ID を持つジョブは順番に処理されますが、異なるグループ ID を持つメッセージは同時に処理できます。

Laravel は、ジョブをディスパッチするときにメッセージグループ ID を指定するためのメソッドチェーンで呼び出せる onGroup メソッドを提供します。

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

メッセージグループを指定せずにジョブを SQS FIFO キューへディスパッチすると、Laravel はキュー名をメッセージグループ ID として使用します。

SQS FIFO キューはメッセージの重複排除をサポートし、1 回だけの処理を保証します。ジョブクラスに deduplicationId メソッドを実装して、カスタム重複排除 ID を提供します。

<?php

namespace App\Jobs;

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

class ProcessSubscriptionRenewal implements ShouldQueue
{
    use Queueable;

    // ...

    /**
     * Get the job's deduplication ID.
     */
    public function deduplicationId(): string
    {
        return "renewal-{$this->subscription->id}";
    }
}

Fair Queues

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;

    // ...

    /**
     * Get the job's message group.
     */
    public function messageGroup(): string
    {
        return "customer-{$this->order->customer_id}";
    }
}

FIFO Listeners, Mail, and Notifications

FIFO キューを使用する場合は、リスナ、メール、通知のメッセージグループも定義する必要があります。あるいは、これらのオブジェクトのキューに入れられたインスタンスを非 FIFO キューにディスパッチすることもできます。

queued event listener のメッセージグループを定義するには、リスナに messageGroup メソッドを定義します。必要に応じて、イベントを受け取り、重複排除 ID を生成するクロージャを返す deduplicator メソッドも定義できます。

<?php

namespace App\Listeners;

use App\Events\OrderShipped;
use Closure;

class SendShipmentNotification
{
    // ...

    /**
     * Get the job's message group.
     */
    public function messageGroup(): string
    {
        return 'shipments';
    }

    /**
     * Get the job's deduplicator.
     */
    public function deduplicator(OrderShipped $event): Closure
    {
        return fn () => "shipment-notification-{$event->order->id}";
    }
}

FIFO キューにキューイングする mail message を送信する場合は、通知の送信時に 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 キューに追加する notification を送信するときは、onGroup メソッドを呼び出してください。必要に応じて、withDeduplicator メソッドも呼び出せます。

use App\Notifications\InvoicePaid;

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

$user->notify($invoicePaid);

Queue Failover

failover キュードライバは、ジョブをキューにプッシュするときに自動フェイルオーバー機能を提供します。 failover 構成のプライマリキュー接続が何らかの理由で失敗した場合、Laravel はリスト内の次の構成された接続にジョブを自動的にプッシュしようとします。これは、キューの信頼性が重要である実稼働環境で高可用性を確保する場合に特に役立ちます。

フェイルオーバーキュー接続を構成するには、failover ドライバを指定し、順番に試行する接続名の配列を指定します。デフォルトでは、Laravel にはアプリケーションの config/queue.php 構成ファイルにサンプルのフェイルオーバー構成が含まれています。

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

failover ドライバを使用する接続を構成したら、フェイルオーバー機能を利用するには、アプリケーションの .env ファイルでフェイルオーバー接続をデフォルトのキュー接続として設定する必要があります。

QUEUE_CONNECTION=failover

次に、フェイルオーバー接続リスト内の接続ごとに少なくとも 1 つのワーカーを開始します。

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 プロセスを実行してください。

Error Handling

ジョブの処理中に例外がスローされた場合、ジョブは自動的にキューに解放され、再試行できるようになります。ジョブは、アプリケーションで許可されている最大回数試行されるまで解放され続けます。最大試行回数は、queue:work Artisan コマンドで使用される --tries スイッチによって定義されます。あるいは、最大試行回数をジョブクラス自体に定義することもできます。キューワーカーの実行に関する詳細は、can be found below を参照してください。

Manually Releasing a Job

場合によっては、ジョブを手動で解放してキューに戻し、後で再試行できるようにしたい場合があります。これを行うには、release メソッドを呼び出します。

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

    $this->release();
}

デフォルトでは、release メソッドはジョブを即時に処理できるようキューへ戻します。ただし、整数または日付インスタンスを release メソッドに渡すことで、指定した秒数が経過するまでジョブを処理できないようにキューに指示できます。

$this->release(10);

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

Manually Failing a Job

場合によっては、ジョブを手動で「失敗」としてマークする必要がある場合があります。これを行うには、fail メソッドを呼び出します。

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

    $this->fail();
}

キャッチした例外のためにジョブを失敗としてマークしたい場合は、例外を fail メソッドに渡すことができます。または、便宜上、例外に変換される文字列エラーメッセージを渡すこともできます。

$this->fail($exception);

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

失敗したジョブの詳細については、documentation on dealing with job failures を参照してください。

Failing Jobs on Specific Exceptions

FailOnException job middleware を使用すると、特定の例外がスローされたときに再試行を即座に中止できます。これにより、外部 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;

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

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        $this->user->authorize('sync-chat-history');

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

        // ...
    }

    /**
     * Get the middleware the job should pass through.
     */
    public function middleware(): array
    {
        return [
            new FailOnException([AuthorizationException::class])
        ];
    }
}

Job Batching

Laravel のジョブバッチ機能を使用すると、ジョブのグループを並行して簡単に実行し、ジョブのバッチの実行が完了したときに何らかのアクションを実行できます。

開始する前に、データベース移行を作成して、完了率などのジョブバッチに関するメタ情報を含むテーブルを構築する必要があります。この移行は、make:queue-batches-table Artisan コマンドを使用して生成できます。

php artisan make:queue-batches-table

php artisan migrate

Defining Batchable Jobs

バッチ可能ジョブを定義するには、通常どおり create a queueable job を実行する必要があります。ただし、ジョブクラスに Illuminate\Bus\Batchable トレイトを追加する必要があります。このトレイトは、ジョブが実行されている現在のバッチを取得するために使用できる 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;

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        if ($this->batch()->cancelled()) {
            // Determine if the batch has been cancelled...

            return;
        }

        // Import a portion of the CSV file...
    }
}

Dispatching Batches

ジョブのバッチをディスパッチするには、Bus ファサードの batch メソッドを使用する必要があります。もちろん、バッチ処理は主に完了コールバックと組み合わせると便利です。したがって、then、catch、および finally メソッドを使用して、バッチの完了コールバックを定義できます。これらの各コールバックは、呼び出されるときに Illuminate\Bus\Batch インスタンスを受け取ります。

複数のキューワーカーを実行する場合、バッチ内のジョブは並行して処理されます。したがって、ジョブが完了する順序は、ジョブがバッチに追加された順序と同じではない場合があります。一連のジョブを順番に実行する方法については、job chains and batches のドキュメントを参照してください。

この例では、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) {
    // The batch has been created but no jobs have been added...
})->progress(function (Batch $batch) {
    // A single job has completed successfully...
})->then(function (Batch $batch) {
    // All jobs completed successfully...
})->catch(function (Batch $batch, Throwable $e) {
    // Batch job failure detected...
})->finally(function (Batch $batch) {
    // The batch has finished executing...
})->dispatch();

return $batch->id;

$batch->id プロパティを介してアクセスできるバッチの ID は、ディスパッチ後のバッチに関する情報を得るために query the Laravel command bus に使用できます。

バッチコールバックはシリアライズされ、Laravel のキューによって後で実行されるため、コールバック内で $this 変数を使用しないでください。また、バッチジョブはデータベーストランザクション内でラップされるため、暗黙的なコミットを発生させるデータベースステートメントをジョブ内で実行しないでください。

Naming Batches

Laravel Horizon や Laravel Telescope などのツールでは、バッチに名前を付けると、バッチに関するよりわかりやすいデバッグ情報を確認できる場合があります。バッチに任意の名前を付けるには、バッチを定義するときに name メソッドを呼び出します。

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // All jobs completed successfully...
})->name('Import CSV')->dispatch();

Batch Connection and Queue

バッチジョブに使用する接続とキューを指定したい場合は、onConnection メソッドと onQueue メソッドを使用できます。すべてのバッチジョブは、同じ接続およびキュー内で実行する必要があります。

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // All jobs completed successfully...
})->onConnection('redis')->onQueue('imports')->dispatch();

Chains and Batches

チェーンされたジョブを配列内に配置することで、バッチ内で chained jobs のセットを定義できます。たとえば、2 つのジョブチェーンを並行して実行し、両方のジョブチェーンの処理が完了したときにコールバックを実行できます。

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) {
    // All jobs completed successfully...
})->dispatch();

逆に、チェーン内でバッチを定義することにより、chain 内でジョブのバッチを実行することもできます。たとえば、最初にジョブのバッチを実行して複数のポッドキャストを公開し、次にジョブのバッチを実行して公開通知を送信できます。

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

Adding Jobs to Batches

場合によっては、バッチ処理されたジョブ内からバッチにジョブを追加すると便利な場合があります。このパターンは、Web リクエスト中にディスパッチするのに時間がかかりすぎる可能性がある数千のジョブをバッチ処理する必要がある場合に役立ちます。したがって、代わりに、バッチにさらに多くのジョブを追加する「ローダー」ジョブの最初のバッチをディスパッチすることもできます。

$batch = Bus::batch([
    new LoadImportBatch,
    new LoadImportBatch,
    new LoadImportBatch,
])->then(function (Batch $batch) {
    // All jobs completed successfully...
})->name('Import Contacts')->dispatch();

この例では、LoadImportBatch ジョブを使用して、追加のジョブでバッチをハイドレートします。これを実現するには、ジョブの batch メソッドを介してアクセスできるバッチインスタンスで add メソッドを使用します。

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

/**
 * Execute the job.
 */
public function handle(): void
{
    if ($this->batch()->cancelled()) {
        return;
    }

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

同じバッチに属するジョブ内からのみ、バッチにジョブを追加できます。

Inspecting Batches

バッチ完了コールバックに提供される Illuminate\Bus\Batch インスタンスには、特定のジョブのバッチの操作と検査を支援するさまざまなプロパティとメソッドがあります。

// The UUID of the batch...
$batch->id;

// The name of the batch (if applicable)...
$batch->name;

// The number of jobs assigned to the batch...
$batch->totalJobs;

// The number of jobs that have not been processed by the queue...
$batch->pendingJobs;

// The number of jobs that have failed...
$batch->failedJobs;

// The number of jobs that have been processed thus far...
$batch->processedJobs();

// The completion percentage of the batch (0-100)...
$batch->progress();

// Indicates if the batch has finished executing...
$batch->finished();

// Cancel the execution of the batch...
$batch->cancel();

// Indicates if the batch has been cancelled...
$batch->cancelled();

Returning Batches From Routes

すべての Illuminate\Bus\Batch インスタンスは JSON シリアル化可能です。つまり、アプリケーションのルートの 1 つから直接インスタンスを返し、完了の進行状況など、バッチに関する情報を含む JSON ペイロードを取得できます。これにより、バッチの完了の進行状況に関する情報をアプリケーションの UI に表示するのが便利になります。

ID でバッチを取得するには、Bus ファサードの findBatch メソッドを使用できます。

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

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

Cancelling Batches

場合によっては、特定のバッチの実行をキャンセルする必要がある場合があります。これは、Illuminate\Bus\Batch インスタンスで cancel メソッドを呼び出すことで実現できます。

/**
 * Execute the job.
 */
public function handle(): void
{
    if ($this->user->exceedsImportLimit()) {
        $this->batch()->cancel();

        return;
    }

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

前の例でお気づきかと思いますが、バッチ処理されたジョブは通常、実行を続行する前に、対応するバッチがキャンセルされたかどうかを判断する必要があります。ただし、便宜上、代わりに SkipIfBatchCancelled middleware をジョブに割り当てることもできます。その名前が示すように、このミドルウェアは、対応するバッチがキャンセルされた場合にジョブを処理しないように Laravel に指示します。

use Illuminate\Queue\Middleware\SkipIfBatchCancelled;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [new SkipIfBatchCancelled];
}

Batch Failures

バッチ処理されたジョブが失敗すると、catch コールバック (割り当てられている場合) が呼び出されます。このコールバックは、バッチ内で失敗した最初のジョブに対してのみ呼び出されます。

Allowing Failures

バッチ内のジョブが失敗すると、Laravel は自動的にバッチを「キャンセル」としてマークします。必要に応じて、ジョブの失敗によってバッチが自動的にキャンセルとしてマークされないように、この動作を無効にすることができます。これは、バッチのディスパッチ中に allowFailures メソッドを呼び出すことで実現できます。

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // All jobs completed successfully...
})->allowFailures()->dispatch();

必要に応じて allowFailures メソッドにクロージャーを渡すことができます。このクロージャーは、ジョブが失敗するたびに実行されます。

$batch = Bus::batch([
    // ...
])->allowFailures(function (Batch $batch, $exception) {
    // Handle individual job failures...
})->dispatch();

Retrying Failed Batch Jobs

便宜上、Laravel には、特定のバッチで失敗したすべてのジョブを簡単に再試行できる queue:retry-batch Artisan コマンドが用意されています。このコマンドは、失敗したジョブを再試行する必要があるバッチの UUID を受け入れます。

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

Pruning Batches

削除処理を行わないと、job_batches テーブルには非常に速いペースでレコードが蓄積します。これを防ぐには、queue:prune-batches Artisan コマンドを毎日実行するよう schedule してください。

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

Storing Batches in DynamoDB

Laravel は、リレーショナルデータベースの代わりに DynamoDB にバッチメタ情報を保存するためのサポートも提供します。ただし、すべてのバッチレコードを保存するには、DynamoDB テーブルを手動で作成する必要があります。

通常、このテーブルには job_batches という名前を付ける必要がありますが、アプリケーションの queue 構成ファイル内の queue.batching.table 構成値の値に基づいてテーブルに名前を付ける必要があります。

DynamoDB Batch Table Configuration

job_batches テーブルには、application という名前の文字列プライマリパーティションキーと、id という名前の文字列プライマリソートキーが必要です。キーの application 部分には、アプリケーションの app 構成ファイル内の name 構成値で定義されたアプリケーションの名前が含まれます。アプリケーション名は DynamoDB テーブルのキーの一部であるため、同じテーブルを使用して複数の Laravel アプリケーションのジョブバッチを保存できます。

さらに、automatic batch pruning を利用したい場合は、テーブルに ttl 属性を定義できます。

DynamoDB Configuration

次に、Laravel アプリケーションが Amazon DynamoDB と通信できるように AWS SDK をインストールします。

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',
],

Pruning Batches in DynamoDB

DynamoDB を使用してジョブバッチ情報を保存する場合、リレーショナルデータベースに保存されているバッチのプルーニングに使用される一般的なプルーニングコマンドは機能しません。代わりに、DynamoDB's native TTL functionality を利用して、古いバッチのレコードを自動的に削除できます。

ttl 属性を使用して DynamoDB テーブルを定義した場合は、設定パラメータを定義して、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 days...
],

Queueing Closures

ジョブクラスをキューにディスパッチする代わりに、クロージャをディスパッチすることもできます。これは、現在のリクエストサイクルの外で実行する必要がある迅速で単純なタスクに最適です。クロージャをキューにディスパッチするとき、クロージャのコードコンテンツは暗号的に署名されるため、転送中に変更することはできません。

use App\Models\Podcast;

$podcast = Podcast::find(1);

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

キューレポートダッシュボードで使用され、queue:work コマンドで表示されるキュークロージャに名前を割り当てるには、name メソッドを使用できます。

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

catch メソッドを使用すると、キューの configured retry attempts をすべて使い果たした後にキューに入れられたクロージャが正常に完了しなかった場合に実行されるクロージャを指定できます。

use Throwable;

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

catch コールバックは Laravel キューによってシリアライズされ、後から実行されるため、catch コールバック内で $this 変数を使用しないでください。

Running the Queue Worker

The `queue:work` Command

Laravel には、キューワーカーを起動し、新しいジョブがキューにプッシュされるときに処理する Artisan コマンドが含まれています。 queue:work Artisan コマンドを使用してワーカーを実行できます。 queue:work コマンドが開始されると、手動で停止するかターミナルを閉じるまで実行が継続されることに注意してください。

php artisan queue:work

queue:work プロセスをバックグラウンドで常時実行するには、Supervisor などのプロセスモニタを使用して、キューワーカーが停止しないようにしてください。

処理されたジョブ ID、接続名、およびキュー名をコマンドの出力に含めたい場合は、queue:work コマンドを呼び出すときに -v フラグを含めることができます。

php artisan queue:work -v

キューワーカーは存続期間の長いプロセスであり、起動されたアプリケーションの状態をメモリに保存することに注意してください。その結果、開始後のコードベースの変更に気付かなくなります。したがって、展開プロセス中は、必ず restart your queue workers を行ってください。さらに、アプリケーションによって作成または変更された静的状態は、ジョブ間で自動的にリセットされないことに注意してください。

あるいは、queue:listen コマンドを実行することもできます。 queue:listen コマンドを使用すると、更新されたコードをリロードしたり、アプリケーションの状態をリセットしたりするときに、ワーカーを手動で再起動する必要がありません。ただし、このコマンドは queue:work コマンドよりも効率が大幅に低くなります。

php artisan queue:listen

Running Multiple Queue Workers

複数のワーカーをキューに割り当ててジョブを同時に処理するには、複数の queue:work プロセスを開始するだけです。これは、ターミナルの複数のタブを使用してローカルで実行することも、プロセスマネージャーの構成設定を使用して運用環境で実行することもできます。 When using Supervisor、numprocs 構成値を使用できます。

Specifying the Connection and Queue

ワーカーがどのキュー接続を使用するかを指定することもできます。 work コマンドに渡される接続名は、config/queue.php 構成ファイルで定義された接続の 1 つに対応する必要があります。

php artisan queue:work redis

デフォルトでは、queue:work コマンドは、指定された接続上のデフォルトキューのジョブのみを処理します。ただし、特定の接続の特定のキューのみを処理することで、キューワーカーをさらにカスタマイズすることもできます。たとえば、すべての電子メールが redis キュー接続の emails キューで処理される場合、次のコマンドを発行して、そのキューのみを処理するワーカーを開始できます。

php artisan queue:work redis --queue=emails

Processing a Specified Number of Jobs

--once オプションを使用すると、キューから 1 つのジョブのみを処理するようにワーカーに指示できます。

php artisan queue:work --once

--max-jobs オプションを使用すると、指定された数のジョブを処理して終了するようにワーカーに指示できます。このオプションは、Supervisor と組み合わせると便利です。これにより、指定された数のジョブの処理後にワーカーが自動的に再起動され、ワーカーが蓄積したメモリが解放されます。

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

Processing All Queued Jobs and Then Exiting

--stop-when-empty オプションを使用すると、すべてのジョブを処理して正常に終了するようにワーカーに指示できます。このオプションは、Docker コンテナ内で Laravel キューを処理するときに、キューが空になった後にコンテナをシャットダウンする場合に便利です。

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

Processing Jobs for a Given Number of Seconds

--max-time オプションを使用すると、指定された秒数の間ジョブを処理してから終了するようにワーカーに指示できます。このオプションは、Supervisor と組み合わせると便利です。これにより、一定時間ジョブを処理した後にワーカーが自動的に再起動され、ワーカーが蓄積したメモリが解放されます。

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

Worker Sleep Duration

ジョブがキューにある場合、ワーカーはジョブ間に遅延なくジョブの処理を続けます。ただし、sleep オプションは、利用可能なジョブがない場合にワーカーが「スリープ」する秒数を決定します。もちろん、ワーカーは寝ている間は新しいジョブを処理しません。

php artisan queue:work --sleep=3

Maintenance Mode and Queues

アプリケーションが maintenance mode の間は、キューに入れられたジョブは処理されません。アプリケーションがメンテナンスモードを終了すると、ジョブは通常どおり処理されます。

メンテナンスモードが有効になっている場合でもキューワーカーにジョブを強制的に処理させるには、--force オプションを使用できます。

php artisan queue:work --force

Resource Considerations

--memory オプションでは、ワーカーが終了するメモリ使用量の上限をメガバイト単位で指定します。デフォルトの上限は 128 メガバイトです。PHP の memory_limit に対する割合も指定できます。

php artisan queue:work --memory=60%

割合による上限を指定するには、PHP の memory_limit が有限である必要があります。PHP のメモリ上限が無制限（-1）の場合は、代わりにメガバイト単位で上限を指定してください。Supervisor などのプロセスモニタを使うと、ワーカーの終了後に自動で再起動できます。

デーモンとして動作するキューワーカーは、各ジョブを処理する前にフレームワークを「再起動」しません。そのため、各ジョブの完了後に負荷の大きいリソースを解放してください。たとえば、image manipulation を GD library で行う場合は、画像の処理が終わった時点で imagedestroy を使ってメモリを解放してください。

Queue Priorities

場合によっては、キューの処理方法に優先順位を付けたい場合があります。たとえば、config/queue.php 構成ファイルで、redis 接続のデフォルトの queue を low に設定できます。ただし、場合によっては、次のようにジョブを high 優先キューにプッシュしたい場合があります。

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

low キュー上のジョブを続行する前に、すべての high キュージョブが処理されたことを検証するワーカーを開始するには、キュー名のカンマ区切りリストを work コマンドに渡します。

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

Queue Workers and Deployment

キューワーカーは存続期間の長いプロセスであるため、再起動されなければコードの変更に気づきません。したがって、キューワーカーを使用してアプリケーションをデプロイする最も簡単な方法は、デプロイメントプロセス中にワーカーを再起動することです。 queue:restart コマンドを発行すると、すべてのワーカーを正常に再起動できます。

php artisan queue:restart

このコマンドは、既存のジョブが失われないように、すべてのキューワーカーに現在のジョブの処理が完了した後に正常に終了するように指示します。 queue:restart コマンドが実行されるとキューワーカーは終了するため、キューワーカーを自動的に再起動するには、Supervisor などのプロセスマネージャーを実行する必要があります。

キューは再起動シグナルの保存に cache を使用するため、この機能を使用する前に、アプリケーションでキャッシュドライバが適切に設定されていることを確認してください。

Reacting to Worker Signals

キューワーカーがジョブの処理中に SIGQUIT、SIGTERM、SIGINT などの終了シグナルを受信すると、ワーカーは現在のジョブの処理を完了してから終了します。ただし、サーバーまたはコンテナーオーケストレーターによってプロセスが停止される前に、ジョブがシグナルに反応する必要がある場合があります。たとえば、長時間実行されるインポートジョブでは、新しいレコードの取得を停止し、現在の進行状況を保存する必要がある場合があります。

ジョブ内からワーカー信号に反応するには、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;

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

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        foreach ($this->import->pendingRows() as $row) {
            if ($this->shouldStop) {
                break;
            }

            // Import the product row...
        }

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

    /**
     * Handle a signal received by the queue worker.
     */
    public function interrupted(int $signal): void
    {
        $this->shouldStop = true;
    }
}

interrupted メソッドは、ジョブの実行中にワーカーがプロセス信号を受信した場合にのみ呼び出されます。これは、timeouts またはジョブの failed method に代わるものではありません。

Job Expirations and Timeouts

Job Expiration

config/queue.php 構成ファイルでは、各キュー接続が retry_after オプションを定義します。このオプションは、処理中のジョブを再試行する前にキュー接続が待機する秒数を指定します。たとえば、retry_after の値が 90 に設定されている場合、ジョブは解放または削除されずに 90 秒間処理されていた場合、キューに解放されます。通常、retry_after 値は、ジョブの処理が完了するまでに合理的にかかる最大秒数に設定する必要があります。

retry_after 値を含まないキュー接続は Amazon SQS だけです。SQS は、AWS コンソールで管理される Default Visibility Timeout に基づいてジョブを再試行します。

Worker Timeouts

queue:work Artisan コマンドは、--timeout オプションを公開します。デフォルトでは、--timeout 値は 60 秒です。ジョブの処理がタイムアウト値で指定された秒数を超えた場合、ジョブを処理しているワーカーはエラーで終了します。通常、ワーカーは process manager configured on your server によって自動的に再起動されます。

php artisan queue:work --timeout=60

retry_after 構成オプションと --timeout CLI オプションは異なりますが、連携してジョブが失われず、ジョブが 1 回だけ正常に処理されるようにします。

--timeout の値は、retry_after の設定値より常に数秒以上短くしてください。これにより、停止したジョブを処理しているワーカーが、ジョブの再試行前に必ず終了します。--timeout オプションが retry_after の設定値より長い場合、ジョブが2回処理される可能性があります。

Pausing and Resuming Queue Workers

場合によっては、キューワーカーを完全に停止せずに、キューワーカーによる新しいジョブの処理を一時的に禁止する必要がある場合があります。たとえば、システムメンテナンス中にジョブの処理を一時停止したい場合があります。 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

すべての接続のすべてのキューでジョブ処理を再開するには、queue:resume コマンドで --all オプションを使用します。

php artisan queue:resume --all

キューの処理を再開すると、ワーカーはそのキューの新しいジョブを直ちに処理し始めます。すべてのキューを再開しても、個別に一時停止したキューは再開されません。なお、キューを一時停止してもワーカープロセス自体は停止しません。指定したキューから新しいジョブを処理しなくなるだけです。

Worker Restart and Pause Signals

デフォルトでは、キューワーカーはジョブの反復ごとにキャッシュドライバをポーリングして再起動および一時停止の信号を確認します。このポーリングは、queue:restart および queue:pause コマンドに応答するために不可欠ですが、パフォーマンスにわずかなオーバーヘッドが発生します。

パフォーマンスを最適化する必要があり、これらの中断機能が必要ない場合は、Queue ファサードで withoutInterruptionPolling メソッドを呼び出して、このポーリングをグローバルに無効にすることができます。これは通常、AppServiceProvider の boot メソッドで実行する必要があります。

use Illuminate\Support\Facades\Queue;

/**
 * Bootstrap any application services.
 */
public function boot(): void
{
    Queue::withoutInterruptionPolling();
}

あるいは、Illuminate\Queue\Worker クラスの静的 $restartable プロパティまたは $pausable プロパティを設定することで、再起動シグナルまたは一時停止シグナルのポーリングを個別に無効にすることもできます。

use Illuminate\Queue\Worker;

/**
 * Bootstrap any application services.
 */
public function boot(): void
{
    Worker::$restartable = false;
    Worker::$pausable = false;
}

割り込みポーリングを無効にすると、ワーカーは無効にした機能に応じて、queue:restart または queue:pause コマンドに応答しなくなります。

Supervisor Configuration

運用環境では、queue:work プロセスを実行し続ける方法が必要です。 queue:work プロセスは、ワーカータイムアウトの超過や queue:restart コマンドの実行など、さまざまな理由で実行を停止することがあります。

このため、queue:work プロセスの終了を検出し、自動的に再起動できるプロセスモニターを構成する必要があります。さらに、プロセスモニターを使用すると、同時に実行する queue:work プロセスの数を指定できます。 Supervisor は Linux 環境で一般的に使用されるプロセスモニターであり、その構成方法については次のドキュメントで説明します。

Installing Supervisor

Supervisorは Linux オペレーティングシステムのプロセスモニターであり、queue:work プロセスが失敗した場合に自動的に再起動します。 Ubuntu に Supervisor をインストールするには、次のコマンドを使用できます。

sudo apt-get install supervisor

Supervisor の設定や管理を自分で行うのが負担に感じられる場合は、Laravel のキューワーカーを実行するためのフルマネージドプラットフォームを提供する Laravel Cloud の利用を検討してください。

Configuring Supervisor

スーパーバイザ設定ファイルは通常、/etc/supervisor/conf.d ディレクトリに保存されます。このディレクトリ内に、スーパーバイザにプロセスの監視方法を指示する構成ファイルをいくつでも作成できます。たとえば、queue:work プロセスを開始および監視する laravel-worker.conf ファイルを作成してみましょう。

[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 ディレクティブは、8 つの queue:work プロセスを実行してすべてを監視し、失敗した場合は自動的に再起動するようにSupervisorに指示します。必要なキュー接続とワーカーオプションを反映するには、構成の command ディレクティブを変更する必要があります。

stopwaitsecs の値は、最も実行時間の長いジョブの処理にかかる秒数より大きくしてください。そうしないと、Supervisor が処理の完了前にジョブを強制終了する可能性があります。

Starting Supervisor

設定ファイルが作成されたら、次のコマンドを使用してスーパーバイザ設定を更新し、プロセスを開始できます。

sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start "laravel-worker:*"

スーパーバイザの詳細については、Supervisor documentation を参照してください。

Dealing With Failed Jobs

キューに投入したジョブが失敗することもあります。心配はいりません。いつも計画どおりに進むとは限らないものです。Laravel には、specify the maximum number of times a job should be attemptedための便利な方法が用意されています。非同期ジョブの試行回数がこの値を超えると、そのジョブは failed_jobs データベーステーブルに登録されます。失敗したSynchronously dispatched jobsはこのテーブルには保存されず、例外はアプリケーションによって直ちに処理されます。

failed_jobs テーブルを作成するための移行は、通常、新しい Laravel アプリケーションにすでに存在しています。ただし、アプリケーションにこのテーブルの移行が含まれていない場合は、make:queue-failed-table コマンドを使用して移行を作成できます。

php artisan make:queue-failed-table

php artisan migrate

queue worker プロセスを実行する場合、queue:work コマンドの --tries スイッチを使用して、ジョブの試行の最大回数を指定できます。 --tries オプションの値を指定しない場合、ジョブは 1 回だけ、またはジョブクラスの 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 メソッドを定義できます。

/**
 * Calculate the number of seconds to wait before retrying the job.
 */
public function backoff(): int
{
    return 3;
}

バックオフ値の配列を定義することで、「指数関数的」バックオフを簡単に構成できます。この例では、再試行の遅延は、最初の再試行では 1 秒、2 回目の再試行では 5 秒、3 回目の再試行では 10 秒、さらに試行が残っている場合はその後の再試行ごとに 10 秒になります。

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Backoff;

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

Cleaning Up After Failed Jobs

特定のジョブが失敗した場合、ユーザーにアラートを送信したり、ジョブによって部分的に完了したアクションを元に戻したりすることができます。これを実現するには、ジョブクラスで 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 のインスタンスになります。

Retrying Failed Jobs

failed_jobs データベーステーブルに挿入された失敗したジョブをすべて表示するには、queue:failed Artisan コマンドを使用できます。

php artisan queue:failed

queue:failed コマンドは、ジョブ ID、接続、キュー、失敗時間、およびジョブに関するその他の情報をリストします。ジョブ ID は、失敗したジョブを再試行するために使用できます。たとえば、ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece の ID を持つ失敗したジョブを再試行するには、次のコマンドを発行します。

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 コマンドを実行し、ID として all を渡します。

php artisan queue:retry all

失敗したジョブを削除したい場合は、queue:forget コマンドを使用できます。

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

Horizon を使用する場合、失敗したジョブを削除するには queue:forget コマンドではなく horizon:forget コマンドを使用してください。

失敗したジョブをすべて failed_jobs テーブルから削除するには、queue:flush コマンドを使用します。

php artisan queue:flush

queue:flush コマンドは、失敗したジョブの古さに関係なく、失敗したすべてのジョブレコードをキューから削除します。 --hours オプションを使用すると、特定の時間前またはそれ以前に失敗したジョブのみを削除できます。

php artisan queue:flush --hours=48

Ignoring Missing Models

Eloquent モデルをジョブに挿入すると、モデルはキューに置かれる前に自動的にシリアル化され、ジョブの処理時にデータベースから再取得されます。ただし、ジョブがワーカーによる処理を待機している間にモデルが削除された場合、ジョブは ModelNotFoundException で失敗する可能性があります。

便宜上、ジョブクラスの DeleteWhenMissingModels 属性を使用して、モデルが欠落しているジョブを自動的に削除することを選択できます。この属性が存在する場合、Laravel は例外を発生させずに静かにジョブを破棄します。

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\DeleteWhenMissingModels;

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

Pruning Failed Jobs

queue:prune-failed Artisan コマンドを呼び出して、アプリケーションの failed_jobs テーブル内のレコードを削除できます。

php artisan queue:prune-failed

デフォルトでは、24 時間を超えて経過した失敗したジョブレコードはすべて削除されます。コマンドに --hours オプションを指定すると、過去 N 時間以内に挿入された失敗したジョブレコードのみが保持されます。たとえば、次のコマンドは、挿入から 48 時間を超えたすべての失敗したジョブレコードを削除します。

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

Storing Failed Jobs in DynamoDB

Laravel は、失敗したジョブレコードをリレーショナルデータベーステーブルではなく DynamoDB に保存するサポートも提供します。ただし、失敗したジョブレコードをすべて保存するには、DynamoDB テーブルを手動で作成する必要があります。通常、このテーブルには failed_jobs という名前を付ける必要がありますが、アプリケーションの queue 構成ファイル内の queue.failed.table 構成値の値に基づいてテーブルに名前を付ける必要があります。

failed_jobs テーブルには、application という名前の文字列プライマリパーティションキーと、uuid という名前の文字列プライマリソートキーが必要です。キーの application 部分には、アプリケーションの app 構成ファイル内の name 構成値で定義されたアプリケーションの名前が含まれます。アプリケーション名は DynamoDB テーブルのキーの一部であるため、同じテーブルを使用して複数の Laravel アプリケーションの失敗したジョブを保存できます。

さらに、Laravel アプリケーションが Amazon DynamoDB と通信できるように、必ず AWS SDK をインストールしてください。

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',
],

Disabling Failed Job Storage

queue.failed.driver 構成オプションの値を null に設定することで、失敗したジョブを保存せずに破棄するように Laravel に指示できます。通常、これは QUEUE_FAILED_DRIVER 環境変数を介して実現できます。

QUEUE_FAILED_DRIVER=null

Failed Job Events

ジョブが失敗したときに呼び出されるイベントリスナを登録したい場合は、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
        });
    }
}

Clearing Jobs From Queues

Horizonを使用している場合は、queue:clear コマンドではなく horizon:clear コマンドを使用してキューからジョブを削除してください。

デフォルト接続のデフォルトキューからすべてのジョブを削除したい場合は、queue:clear Artisan コマンドを使用して削除できます。

php artisan queue:clear

connection 引数と queue オプションを指定して、特定の接続とキューからジョブを削除することもできます。

php artisan queue:clear redis --queue=emails

キューからジョブを削除できるのは、SQS、Redis、データベースのキュードライバを使用している場合だけです。また、SQS のメッセージ削除処理には最大 60 秒かかるため、キューをクリアしてから 60 秒以内に SQS キューへ送信されたジョブも削除される可能性があります。

Monitoring Your Queues

キューにジョブが突然殺到すると、キューが過剰になり、ジョブが完了するまでの待ち時間が長くなる可能性があります。必要に応じて、Laravel はキューのジョブ数が指定されたしきい値を超えたときに警告を発することができます。

まず、queue:monitor コマンドをrun every minuteするようスケジュールしてください。このコマンドでは、監視したいキューの名前と、設定したいジョブ数のしきい値を指定できます。

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

Testing

ジョブをディスパッチするコードをテストする場合、ジョブのコードは、それをディスパッチするコードとは別に直接テストできるため、実際にはジョブ自体を実行しないように 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 メソッドにクロージャーを渡すことができます。指定された条件を満たすジョブが少なくとも 1 つプッシュされた場合、アサーションは成功します。

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 の第2引数として、あるいは assertPushedOn の第3引数として渡すこともできます。

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

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

指定したすべてのプロパティが一致する必要があります。値は厳密比較され、Eloquent モデルは is メソッドで比較されます。また、Bus ファサードの assertDispatched、assertNotDispatched、assertDispatchedSync、assertDispatchedAfterResponse メソッドも、期待するプロパティ値の配列を第2引数として受け取ります。

Faking a Subset of Jobs

他のジョブを通常どおり実行できるようにしながら、特定のジョブのみを偽装する必要がある場合は、偽装するジョブのクラス名を fake メソッドに渡すことができます。

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

    // Perform order shipping...

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

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

    // Perform order shipping...

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

except メソッドを使用すると、指定したジョブのセットを除くすべてのジョブを偽装できます。

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

Testing Job Chains

ジョブチェーンをテストするには、Bus ファサードのフェイク機能を利用します。Bus ファサードの assertChained メソッドを使用すると、chain of jobs がディスパッチされたことをアサートできます。assertChained メソッドは、チェーンされたジョブの配列を第1引数として受け取ります。

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

Testing Chain Modifications

チェーンされたジョブが既存のチェーンの先頭または末尾にジョブを追加する場合（prepends or appends jobs to an existing chain）、ジョブの assertHasChain メソッドを使用して、ジョブに予想される残りのジョブのチェーンがあることをアサートできます。

$job = new ProcessPodcast;

$job->handle();

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

assertDoesntHaveChain メソッドを使用して、ジョブの残りのチェーンが空であることをアサートできます。

$job->assertDoesntHaveChain();

Testing Chained Batches

ジョブチェーンにジョブのバッチが含まれている場合（contains a batch of jobs）、チェーンアサーション内に 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,
]);

Testing Job Batches

Bus ファサードの assertBatched メソッドを使用すると、batch of jobs がディスパッチされたことをアサートできます。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();

Testing Job / Batch Interaction

さらに、場合によっては、個々のジョブとその基礎となるバッチとの対話をテストする必要があるかもしれません。たとえば、ジョブがそのバッチのさらなる処理をキャンセルしたかどうかをテストする必要がある場合があります。これを実現するには、withFakeBatch メソッドを介してジョブに偽のバッチを割り当てる必要があります。 withFakeBatch メソッドは、ジョブインスタンスと偽のバッチを含むタプルを返します。

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

$job->handle();

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

Testing Job / Queue Interactions

場合によっては、キューに入れられたジョブが自らキューに戻るかどうか（releases itself back onto the queue）をテストする必要があるかもしれません。または、ジョブ自体が削除されたかどうかをテストする必要がある場合があります。ジョブをインスタンス化し、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();

Job Events

Queue facade の before メソッドと after メソッドを使用すると、キューに入れたジョブの処理前または処理後に実行するコールバックを指定できます。これらのコールバックは、追加のログを記録したり、ダッシュボード用の統計情報を増分したりするのに適しています。通常、これらのメソッドは service provider の 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
{
    /**
     * Register any application services.
     */
    public function register(): void
    {
        // ...
    }

    /**
     * Bootstrap any application services.
     */
    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 facade の 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
});

Introduction​

Connections vs. Queues​

Driver Notes and Prerequisites​

Database​

Redis​

Redis Cluster​

Blocking​

SQS Overflow Storage​

Other Driver Prerequisites​

Creating Jobs​

Generating Job Classes​

Class Structure​

handle Method Dependency Injection​

Queued Relationships​

Unique Jobs​

Keeping Jobs Unique Until Processing Begins​

Unique Job Locks​

Debounced Jobs​

Encrypted Jobs​

Job Middleware​

Rate Limiting​

Rate Limiting With Redis​

Preventing Job Overlaps​

Sharing Lock Keys Across Job Classes​

Throttling Exceptions​

Throttling Exceptions With Redis​

Releasing Jobs​

Skipping Jobs​

Dispatching Jobs​

Delayed Dispatching​

Synchronous Dispatching​

Deferred Dispatching​

Bulk Dispatching​

Preparing Jobs Before Dispatch​

Jobs & Database Transactions​

Specifying Commit Dispatch Behavior Inline​

Job Chaining​

Chain Connection and Queue​

Adding Jobs to the Chain​

Chain Failures​

Customizing the Queue and Connection​

Dispatching to a Particular Queue​

Dispatching to a Particular Connection​

Queue Routing​

Specifying Max Job Attempts / Timeout Values​

Max Attempts​

Time Based Attempts​

Max Exceptions​

Stopping Retries by Exception​

Timeout​

Failing on Timeout​

SQS FIFO and Fair Queues​

Fair Queues​

FIFO Listeners, Mail, and Notifications​

Queue Failover​

Error Handling​

Manually Releasing a Job​

Manually Failing a Job​

Failing Jobs on Specific Exceptions​

Job Batching​

Defining Batchable Jobs​

Dispatching Batches​

Naming Batches​

Batch Connection and Queue​

Chains and Batches​

Adding Jobs to Batches​

Inspecting Batches​

Returning Batches From Routes​

Cancelling Batches​

Batch Failures​

Allowing Failures​

Retrying Failed Batch Jobs​

Pruning Batches​

Storing Batches in DynamoDB​

DynamoDB Batch Table Configuration​

DynamoDB Configuration​

Pruning Batches in DynamoDB​

Queueing Closures​

Running the Queue Worker​

The queue:work Command​

Introduction

Connections vs. Queues

Driver Notes and Prerequisites

Database

Redis

Redis Cluster

Blocking

SQS Overflow Storage

Other Driver Prerequisites

Creating Jobs

Generating Job Classes

Class Structure

`handle` Method Dependency Injection

Queued Relationships

Unique Jobs

Keeping Jobs Unique Until Processing Begins

Unique Job Locks

Debounced Jobs

Encrypted Jobs

Job Middleware

Rate Limiting

Rate Limiting With Redis

Preventing Job Overlaps

Sharing Lock Keys Across Job Classes

Throttling Exceptions

Throttling Exceptions With Redis

Releasing Jobs

Skipping Jobs

Dispatching Jobs

Delayed Dispatching

Synchronous Dispatching

Deferred Dispatching

Bulk Dispatching

Preparing Jobs Before Dispatch

Jobs & Database Transactions

Specifying Commit Dispatch Behavior Inline

Job Chaining

Chain Connection and Queue

Adding Jobs to the Chain

Chain Failures

Customizing the Queue and Connection

Dispatching to a Particular Queue

Dispatching to a Particular Connection

Queue Routing

Specifying Max Job Attempts / Timeout Values

Max Attempts

Time Based Attempts

Max Exceptions

Stopping Retries by Exception

Timeout

Failing on Timeout

SQS FIFO and Fair Queues

Fair Queues

FIFO Listeners, Mail, and Notifications

Queue Failover

Error Handling

Manually Releasing a Job

Manually Failing a Job

Failing Jobs on Specific Exceptions

Job Batching

Defining Batchable Jobs

Dispatching Batches

Naming Batches

Batch Connection and Queue

Chains and Batches

Adding Jobs to Batches

Inspecting Batches

Returning Batches From Routes

Cancelling Batches

Batch Failures

Allowing Failures

Retrying Failed Batch Jobs

Pruning Batches

Storing Batches in DynamoDB

DynamoDB Batch Table Configuration

DynamoDB Configuration

Pruning Batches in DynamoDB

Queueing Closures

Running the Queue Worker

The `queue:work` Command