キュー 7.x Laravel

イントロダクション

Tip!! 現在、LaravelはRedisで動作するキューのための美しいダッシュボードと設定システムを備えたHorizonを提供しています。詳細は、Horizonのドキュメントで確認してください。

Laravelのキューサービスは、Beanstalk、Amazon SQS、Redis、さらにはリレーショナル・データベースなどさまざまなキューバックエンドに対し共通のAPIを提供しています。キューによりメール送信のような時間を費やす処理を遅らせることが可能です。時間のかかるタスクを遅らせることで、よりアプリケーションのリクエストをドラマチックにスピードアップできます。

キューの設定ファイルはconfig/queue.phpです。このファイルにはフレームワークに含まれているそれぞれのドライバーへの接続設定が含まれています。それにはデータベース、Beanstalkd、Amazon SQS、Redis、ジョブが即時に実行される同期（ローカル用途）ドライバーが含まれています。 nullキュードライバはキューされたジョブが実行されないように、破棄します。

接続 Vs. キュー

Laravelのキューへ取り掛かる前に、「接続」と「キュー」の区別を理解しておくことが重要です。config/queue.php設定ファイルの中には、connections設定オプションがあります。このオプションはAmazon SQS、Beanstalk、Redisなどのバックエンドサービスへの個々の接続を定義します。しかし、どんな指定されたキュー接続も、複数の「キュー」を持つことができます。「キュー」とはキュー済みのジョブのスタック、もしくは積み重ねのことです。

queue接続ファイルのqueue属性を含んでいる、各接続設定例に注目してください。ジョブがディスパッチされ、指定された接続へ送られた時にのデフォルトキューです。言い換えれば、どのキューへディスパッチするのか明確に定義していないジョブをディスパッチすると、そのジョブは接続設定のqueue属性で定義したキューへ送られます。

// このジョブはデフォルトキューへ送られる
Job::dispatch();

// このジョブは"emails"キューへ送られる
Job::dispatch()->onQueue('emails');

あるアプリケーションでは複数のキューへジョブを送る必要はなく、代わりに１つのシンプルなキューが適しているでしょう。しかし、複数のキューへジョブを送ることは優先順位づけしたい、もしくはジョブの処理を分割したいアプリケーションでとくに便利です。Laravelのキューワーカはプライオリティによりどのキューで処理するかを指定できるからです。たとえば、ジョブをhighキューへ送れば、より高い処理プライオリティのワーカを実行できます。

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

ドライバの注意事項と要件

データベース

databaseキュードライバを使用するには、ジョブを記録するためのデータベーステーブルが必要です。このテーブルを作成するマイグレーションはqueue:table Artisanコマンドにより生成できます。マイグレーションが生成されたら、migrateコマンドでデータベースをマイグレートしてください。

php artisan queue:table

php artisan migrate

Redis

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

Redisクラスタ

Redisキュー接続でRedisクラスタを使用している場合は、キュー名にキーハッシュタグを含める必要があります。これはキューに指定した全Redisキーが同じハッシュスロットに確実に置かれるようにするためです。

'redis' => [
    'driver' => 'redis',
    'connection' => 'default',
    'queue' => '{default}',
    'retry_after' => 90,
],

ブロッキング

Redisキューを使用する場合、ワーカのループの繰り返しとRedisデータベースに対する再ポールの前に、ジョブを実行可能にするまでどの程度待つのかを指定する、block_for設定オプションを使うことができます。

新しいジョブを得るため、Redisデータベースに連続してポールしてしまうより、キューの負荷にもとづきより効率的になるよう、この値を調整してください。たとえば、ジョブを実行可能にするまで、ドライバーが５秒間ブロックするように指示するには、値に5をセットします。

'redis' => [
    'driver' => 'redis',
    'connection' => 'default',
    'queue' => 'default',
    'retry_after' => 90,
    'block_for' => 5,
],

Note: block_forへ0を設定するとジョブが利用可能になるまで、キューワーカを無制限にブロックしてしまいます。これはさらに、次のジョブが処理されるまで、SIGTERMのようなシグナルが処理されるのも邪魔してしまいます。

他のドライバの要件

以下の依存パッケージがリストしたキュードライバを使用するために必要です。

Amazon SQS: aws/aws-sdk-php ~3.0
Beanstalkd: pda/pheanstalk ~4.0
Redis: predis/predis ~1.0、もしくはphpredis PHP拡張

ジョブの作成

ジョブクラスの生成

キュー投入可能なアプリケーションの全ジョブは、デフォルトでapp/Jobsディレクトリへ保存されます。app/Jobsディレクトリが存在しなくても、make:job Artisanコマンドの実行時に生成されます。新しいキュージョブをArtisan CLIで生成できます。

php artisan make:job ProcessPodcast

非同期で実行するため、ジョブをキューへ投入することをLaravelへ知らせる、Illuminate\Contracts\Queue\ShouldQueueインターフェイスが生成されたクラスには実装されます。

Tip!! Job stubs may be customized using stub publishing

クラス構成

ジョブクラスは通常とてもシンプルで、キューによりジョブが処理される時に呼び出される、handleメソッドのみで構成されています。手始めに、ジョブクラスのサンプルを見てみましょう。この例は、ポッドキャストの公開サービスを管理し、公開前にアップロードしたポッドキャストファイルを処理する必要があるという仮定です。

<?php

namespace App\Jobs;

use App\AudioProcessor;
use App\Podcast;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

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

    protected $podcast;

    /**
     * 新しいジョブインスタンスの生成
     *
     * @param  Podcast  $podcast
     * @return void
     */
    public function __construct(Podcast $podcast)
    {
        $this->podcast = $podcast;
    }

    /**
     * ジョブの実行
     *
     * @param  AudioProcessor  $processor
     * @return void
     */
    public function handle(AudioProcessor $processor)
    {
        // アップロード済みポッドキャストの処理…
    }
}

この例中、キュージョブのコンテナーに直接Eloquentモデルが渡せることに注目してください。ジョブが使用しているSerializesModelsトレイトによりEloquentモデルとロード済みのリレーションは優雅にシリアライズされ、ジョブが処理される時にアンシリアライズされます。キュー投入されたジョブがコンテナでEloquentモデルを受け取ると、モデルの識別子のみシリアライズされています。ジョブが実際に処理される時、キューシステムは自動的にデータベースから完全なモデルインスタンスとロード済みだったリレーションを再取得します。これらはすべてアプリケーションの完全な透過性のためであり、Eloquentモデルインスタンスをシリアライズするときに発生する問題を防ぐことができます。

handleメソッドはキューによりジョブが処理されるときに呼びだされます。ジョブのhandleメソッドにタイプヒントにより依存を指定できることに注目してください。Laravelのサービスコンテナが自動的に依存を注入します。

もし、どのようにコンテナが依存をhandleメソッドへ注入するかを完全にコントロールしたい場合は、コンテナのbindMethodメソッドを使用します。bindMethodメソッドは、ジョブとコンテナを受け取るコールバックを引数にします。コールバックの中で、お望みのまま自由にhandleメソッドを起動できます。通常は、サービスプロバイダからこのメソッドを呼び出すべきでしょう。

use App\Jobs\ProcessPodcast;

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

Note: Rawイメージコンテンツのようなバイナリデータは、キュージョブへ渡す前に、base64_encode関数を通してください。そうしないと、そのジョブはキューへ設置する前にJSONへ正しくシリアライズされません。

リレーションの処理

ロード済みのリレーションもシリアライズされるため、シリアライズ済みのジョブ文字列は極めて大きくなり得ます。リレーションがシリアライズされるのを防ぐには、プロパティの値を設定するときにモデルのwithoutRelationsメソッドを呼び出してください。このメソッドは、ロード済みのリレーションを外したモデルのインスタンスを返します。

/**
 * 新しいジョブインスタンスの生成
 *
 * @param  \App\Podcast  $podcast
 * @return void
 */
public function __construct(Podcast $podcast)
{
    $this->podcast = $podcast->withoutRelations();
}

ジョブミドルウェア

ジョブミドルウェアはキュー済みジョブの実行周りのカスタムロジックをラップできるようにし、ジョブ自身の定形コードを減らします。例として、５分毎に１ジョブのみを処理するために、LaravelのRedisレート制限機能を活用する、以下のhandleメソッドを考えてみましょう。

/**
 * ジョブの実行
 *
 * @return void
 */
public function handle()
{
    Redis::throttle('key')->block(0)->allow(1)->every(5)->then(function () {
        info('Lock obtained...');

        // ジョブの処理…
    }, function () {
        // ロック取得ができない…

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

このコードは有効ですが、Redisレート制限ロジックが散らかっているため、handleメソッドの構造はうるさくなりました。さらに、レート制限をかけたい他のジョブでもこのレート制限ロジックが重複してしまいます。

handleメソッドの中でレート制限をする代わりに、レート制限を処理するジョブミドルウェアを定義できます。Laravelはジョブミドルウェアの置き場所を決めていないため、アプリケーションのどこにでもジョブミドルウェアを設置できます。この例では、app/Jobs/Middlewareディレクトリへミドルウェアを設置しています。

<?php

namespace App\Jobs\Middleware;

use Illuminate\Support\Facades\Redis;

class RateLimited
{
    /**
     * キュー済みジョブの処理
     *
     * @param  mixed  $job
     * @param  callable  $next
     * @return mixed
     */
    public function handle($job, $next)
    {
        Redis::throttle('key')
                ->block(0)->allow(1)->every(5)
                ->then(function () use ($job, $next) {
                    // ロックを取得した場合の処理…

                    $next($job);
                }, function () use ($job) {
                    // ロックを取得できなかった処理…

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

ご覧の通り、ルートミドルウェアと同様に、ジョブミドルウェアも処理するジョブを受け取り、コールバックは処理を続けるため呼び出されます。

ジョブミドルウェアを作成したら、ジョブのmiddlewareメソッドから返すことにより、指定します。このメソッドはジョブのスカフォールドを行うmake:job Artisanコマンドでは作成されないため、ジョブクラスの定義に自身で追加してください。

use App\Jobs\Middleware\RateLimited;

/**
 * このジョブが通過する必要のあるミドルウェアの取得
 *
 * @return array
 */
public function middleware()
{
    return [new RateLimited];
}

ジョブのディスパッチ

ジョブクラスを書き上げたら、ジョブクラス自身のdispatchメソッドを使い、ディスパッチできます。dispatchメソッドへ渡す引数は、ジョブのコンストラクタへ渡されます。

<?php

namespace App\Http\Controllers;

use App\Http\Controllers\Controller;
use App\Jobs\ProcessPodcast;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 新ポッドキャストの保存
     *
     * @param  Request  $request
     * @return Response
     */
    public function store(Request $request)
    {
        // ポッドキャスト作成…

        ProcessPodcast::dispatch($podcast);
    }
}

条件によりジョブをディスパッチする場合は、dispatchIfかdispatchUnlessを使います。

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

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

遅延ディスパッチ

キュー投入されたジョブの実行を遅らせたい場合は、ジョブのディスパッチ時にdelayメソッドを使います。例として、ディスパッチ後１０分経つまでは、処理が行われないジョブを指定してみましょう。

<?php

namespace App\Http\Controllers;

use App\Http\Controllers\Controller;
use App\Jobs\ProcessPodcast;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 新ポッドキャストの保存
     *
     * @param  Request  $request
     * @return Response
     */
    public function store(Request $request)
    {
        // ポッドキャスト作成…

        ProcessPodcast::dispatch($podcast)
                ->delay(now()->addMinutes(10));
    }
}

Note: Amazon SQSキューサービスは、最大１５分の遅延時間です。

レスポンスをブラウザへ送信後のディスパッチ

別の方法として、ユーザーのブラウザにレスポンスを送り終えるまで、ジョブのディスパッチを遅らせるdispatchAfterResponseメソッドがあります。これによりキューされたジョブがまだ実行中であっても、ユーザーはアプリケーションをすぐ使い始めることができます。この方法は通常、メール送信のようなユーザーを数秒待たせるジョブにのみ使うべきでしょう。

use App\Jobs\SendNotification;

SendNotification::dispatchAfterResponse();

dispatchでクロージャをディスパッチし、afterResponseメソッドをチェーンすることで、ブラウザにレスポンスを送り終えたらクロージャを実行することも可能です。

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

dispatch(function () {
    Mail::to('taylor@laravel.com')->send(new WelcomeMessage);
})->afterResponse();

同期ディスパッチ

ジョブを即時（同期的）にディスパッチしたい場合は、dispatchNowメソッドを使用します。このメソッドを使用する場合、そのジョブはキューされずに現在のプロセスで即時実行されます。

<?php

namespace App\Http\Controllers;

use App\Http\Controllers\Controller;
use App\Jobs\ProcessPodcast;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 新ポッドキャストの保存
     *
     * @param  Request  $request
     * @return Response
     */
    public function store(Request $request)
    {
        // ポッドキャスト作成…

        ProcessPodcast::dispatchNow($podcast);
    }
}

ジョブチェーン

主要なジョブが正しく実行し終えた後に連続して実行する必要がある、キュー投入ジョブのリストをジョブチェーンで指定できます。一連のジョブの内、あるジョブが失敗すると、残りのジョブは実行されません。キュー投入ジョブチェーンを実行するには、dispatchableジョブどれかに対し、withChainメソッドを使用します。

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

ジョブクラスインスタンスのチェーンだけでなく、クロージャもチェーンできます。

ProcessPodcast::withChain([
    new OptimizePodcast,
    new ReleasePodcast,
    function () {
        Podcast::update(...);
    },
])->dispatch();

Note: ジョブの削除に$this->delete()メソッドを使用しても、チェーンしたジョブの処理を停止できません。チェーンの実行を停止するのは、チェーン中のジョブが失敗した場合のみです。

チェーンの接続とキュー

ジョブチェーンで使用するデフォルトの接続とキューを指定したい場合は、allOnConnectionとallOnQueueメソッドを使用します。これらのメソッドは、キューされたジョブへ別の接続／キューが明確に指定されていない限り使用される、接続とキューを設定します。

ProcessPodcast::withChain([
    new OptimizePodcast,
    new ReleasePodcast
])->dispatch()->allOnConnection('redis')->allOnQueue('podcasts');

キューと接続のカスタマイズ

特定キューへのディスパッチ

ジョブを異なるキューへ投入することで「カテゴライズ」できますし、さまざまなキューにいくつのワーカを割り当てるかと言うプライオリティ付けもできます。これはキー設定ファイルで定義した、別々のキュー「接続」へのジョブ投入を意味してはいないことに気をつけてください。一つの接続内の複数のキューを指定する方法です。

<?php

namespace App\Http\Controllers;

use App\Http\Controllers\Controller;
use App\Jobs\ProcessPodcast;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 新ポッドキャストの保存
     *
     * @param  Request  $request
     * @return Response
     */
    public function store(Request $request)
    {
        // ポッドキャスト作成…

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

特定の接続へのディスパッチ

複数のキュー接続を利用するなら、ジョブを投入するキューを指定できます。ジョブをディスパッチする時に、onConnectionメソッドで接続を指定します。

<?php

namespace App\Http\Controllers;

use App\Http\Controllers\Controller;
use App\Jobs\ProcessPodcast;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * 新ポッドキャストの保存
     *
     * @param  Request  $request
     * @return Response
     */
    public function store(Request $request)
    {
        // ポッドキャスト作成…

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

ジョブを投入する接続とキューを指定するために、onConnectionとonQueueメソッドをチェーンすることもできます。

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

最大試行回数／タイムアウト値の指定

最大試行回数

ジョブが試行する最大回数を指定するアプローチの一つは、Artisanコマンドラインへ--triesスイッチ使う方法です。

php artisan queue:work --tries=3

しかし、より粒度の高いアプローチは、ジョブクラス自身に最大試行回数を定義する方法です。これはコマンドラインで指定された値より、優先度が高くなっています。

<?php

namespace App\Jobs;

class ProcessPodcast implements ShouldQueue
{
    /**
     * 最大試行回数
     *
     * @var int
     */
    public $tries = 5;
}

時間ベースの試行

失敗するまでジョブの試行を何度認めるかを定義する代わりに、ジョブのタイムアウト時間を定義することもできます。これにより、指定した時間内で複数回ジョブを試行します。タイムアウト時間を定義するには、ジョブクラスにretryUntilメソッドを追加します。

/**
 * タイムアウトになる時間を決定
 *
 * @return \DateTime
 */
public function retryUntil()
{
    return now()->addSeconds(5);
}

Tip!! キューイベントリスナでも、retryUntilメソッドを定義できます。

Max例外

ジョブを何度も再試行するように指定している場合、指定した回数の例外が発生したことをきっかけにしてその再試行を失敗として取り扱いたい場合も起きると思います。そうするにはジョブクラスにmaxExceptionsプロパティを定義してください。

<?php

namespace App\Jobs;

class ProcessPodcast implements ShouldQueue
{
    /**
     * 最大試行回数
     *
     * @var int
     */
    public $tries = 25;

    /**
     * 失敗と判定するまで許す最大例外数
     *
     * @var int
     */
    public $maxExceptions = 3;

    /**
     * ジョブの実行
     *
     * @return void
     */
    public function handle()
    {
        Redis::throttle('key')->allow(10)->every(60)->then(function () {
            // ロックが取得でき、ポッドキャストの処理を行う…
        }, function () {
            // ロックが取得できなかった
            return $this->release(10);
        });
    }
}

この例の場合、アプリケーションがRedisのロックを取得できない場合は、そのジョブは１０秒でリリースされます。そして、２５回再試行を継続します。しかし発生した例外を３回処理しなかった場合、ジョブは失敗します。

タイムアウト

Note: ジョブのタイムアウトを利用するには、pcntlPHP拡張をインストールする必要があります。

同様に、ジョブの最大実行秒数を指定するために、Artisanコマンドラインに--timeoutスイッチを指定できます。

php artisan queue:work --timeout=30

しかしながら、最大実行秒数をジョブクラス自身に定義することもできます。ジョブにタイムアウト時間を指定すると、コマンドラインに指定されたタイムアウトよりも優先されます。

<?php

namespace App\Jobs;

class ProcessPodcast implements ShouldQueue
{
    /**
     * ジョブがタイムアウトになるまでの秒数
     *
     * @var int
     */
    public $timeout = 120;
}

レート制限

Note: この機能が動作するには、アプリケーションでRedisサーバが利用できる必要があります。

アプリケーションでRedisを利用しているなら、時間と回数により、キュージョブを制限できます。この機能は、キュージョブがレート制限のあるAPIに関連している場合に役立ちます。

throttleメソッドの使用例として、指定したジョブタイプを６０秒毎に１０回だけ実行できるように制限しましょう。ロックできなかった場合、あとで再試行できるように、通常はジョブをキューへ戻す必要があります。

Redis::throttle('key')->allow(10)->every(60)->then(function () {
    // ジョブのロジック処理…
}, function () {
    // ロックできなかった場合の処理…

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

Tip!! 上記の例でkeyは、レート制限したいジョブのタイプを表す一意の認識文字列です。たとえば、ジョブのクラス名と（そのジョブに含まれているならば）EloquentモデルのIDを元に、制限できます。

Note: レート制限に引っかかったジョブをキューへ戻す(release)する場合も、ジョブの総試行回数(attempts)は増加します。

もしくは、ジョブを同時に処理するワーカの最大数を指定可能です。これは、一度に一つのジョブが更新すべきリソースを変更するキュージョブを使用する場合に、役立ちます。funnelメソッドの使用例として、一度に１ワーカのみにより処理される、特定のタイプのジョブを制限してみましょう。

Redis::funnel('key')->limit(1)->then(function () {
    // ジョブのロジック処理…
}, function () {
    // ロックできなかった場合の処理…

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

Tip!! レート制限を使用する場合、実行を成功するまでに必要な試行回数を決めるのは、難しくなります。そのため、レート制限は時間ベースの試行と組み合わせるのが便利です。

エラー処理

ジョブの処理中に例外が投げられると、ジョブは自動的にキューへ戻され、再試行されます。ジョブはアプリケーションが許している最大試行回数に達するまで、連続して実行されます。最大試行回数はqueue:work Artisanコマンドへ--triesスイッチを使い定義されます。もしくは、ジョブクラス自身に最大試行回数を定義することもできます。キューワーカの実行についての情報は、以降で説明します。

クロージャのキュー投入

ジョブクラスをキューへディスパッチする代わりに、クロージャもディスパッチできます。これは現在のリクエストサイクル外で実行する必要のある、シンプルなタスクを扱うのに適しています。

$podcast = App\Podcast::find(1);

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

クロージャをキューへディスパッチすると、処理中に改変されないように、クロージャのコード内容は暗号化署名されます。

キューワーカの実行

Laravelには、キューに投入された新しいジョブを処理する、キューワーカも含まれています。queue:work Artisanコマンドを使いワーカを実行できます。queue:workコマンドが起動したら、皆さんが停止するか、ターミナルを閉じるまで実行し続けることに注意してください。

php artisan queue:work

Tip!! バックグランドでqueue:workプロセスを永続的に実行し続けるには、キューワーカが止まらずに実行し続けていることを確実にするため、Supervisorのようなプロセスモニタを利用する必要があります。

キューワーカは長時間起動するプロセスで、メモリ上にアプリケーション起動時の状態を保存していることを記憶にとどめてください。そのため、開発段階ではキューワーカの再起動を確実に実行してください。付け加えて、アプリケーションにより生成、もしくは変更された静的な状態は、ジョブ間で自動的にリセットされないことも覚えておきましょう。

別の方法として、queue:listenコマンドを実行することもできます。queue:listenコマンドを使えば更新したコードをリロード、もしくはアプリケーションの状態をリセットしたい場合に、手動でワーカをリスタートする必要がなくなります。しかし、このコマンドはqueue:workほど効率はよくありません。

php artisan queue:listen

接続とキューの指定

どのキュー接続をワーカが使用するのかを指定できます。workコマンドで指定する接続名は、config/queue.php設定ファイルで定義されている接続と対応します。

php artisan queue:work redis

指定した接続の特定のキューだけを処理するように、さらにキューワーカをカスタマイズすることもできます。たとえば、メールの処理をすべて、redisキュー接続のemailsキューで処理する場合、以下のコマンドでキューの処理だけを行うワーカを起動できます。

php artisan queue:work redis --queue=emails

ジョブを一つ処理する

--onceオプションは、ワーカにキュー中のジョブをひとつだけ処理するように指示します。

php artisan queue:work --once

キューされたすべてのジョブを処理し、終了する

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

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

リソースの考察

デーモンキューワーカは各ジョブを処理する前に、フレームワークを「再起動」しません。そのため、各ジョブが終了したら、大きなリソースを開放してください。たとえば、GDライブラリでイメージ処理を行ったら、終了前にimagedestroyにより、メモリを開放してください。

キュープライオリティ

ときどき、キューをどのように処理するかをプライオリティ付けしたいことも起きます。たとえば、config/queue.phpでredis接続のデフォルトqueueをlowに設定したとしましょう。しかし、あるジョブをhighプライオリティでキューへ投入したい場合です。

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

lowキュー上のジョブの処理が継続される前に、全highキュージョブが処理されることを確実にするには、workコマンドのキュー名にコンマ区切りのリストで指定してください。

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

キューワーカとデプロイ

キューワーカは長時間起動プロセスであるため、リスタートしない限りコードの変更を反映しません。ですから、キューワーカを使用しているアプリケーションをデプロイする一番シンプルな方法は、デプロイ処理の間、ワーカをリスタートすることです。queue:restartコマンドを実行することで、全ワーカを穏やかに再起動できます。

php artisan queue:restart

このコマンドは存在しているジョブが失われないように、現在のジョブの処理が終了した後に、全キューワーカーへ穏やかに「終了する(die)」よう指示します。キューワーカはqueue:restartコマンドが実行されると、終了するわけですから、キュージョブを自動的に再起動する、Supervisorのようなプロセスマネージャーを実行すべきでしょう。

Tip!! このコマンドはリスタートシグナルを保存するために、キャッシュを使用します。そのため、この機能を使用する前に、アプリケーションのキャッシュドライバーが、正しく設定されていることを確認してください。

ジョブの期限切れとタイムアウト

ジョブの有効期限

config/queue.php設定ファイルの中で、各キュー接続はretry_afterオプションを定義しています。このオプションは処理中のジョブを再試行するまで、キュー接続を何秒待つかを指定します。たとえば、retry_afterの値が90であれば、そのジョブは９０秒の間に削除されることなく処理され続ければ、キューへ再投入されます。通常、retry_after値はジョブが処理を妥当に完了するまでかかるであろう秒数の最大値を指定します。

Note: retry_afterを含まない唯一の接続は、Amazon SQSです。SQSはAWSコンソールで管理する、Default Visibility Timeoutを元にリトライを行います。

ワーカタイムアウト

queue:work Artisanコマンドは--timeoutオプションも提供しています。--timeoutオプションはLaravelキューマスタプロセスが、ジョブを処理する子のキューワーカをKillするまでどのくらい待つかを指定します。さまざまな理由により、時に子のキュープロセスは「フリーズ」します。--timeoutオプションは、指定した実行時間を過ぎたフリーズプロセスを取り除きます。

php artisan queue:work --timeout=60

retry_after設定オプションと--timeout CLIオプションは異なります。しかし、確実にジョブを失わずに、一度だけ処理を完了できるよう共に働きます。

Note: --timeout値は、最低でも数秒retry_after設定値よりも短くしてください。これにより、与えられたジョブを処理するワーカが、ジョブのリトライ前に確実にkillされます。--timeoutオプションをretry_after設定値よりも長くすると、ジョブが２度実行されるでしょう。

ワーカスリープ時間

ジョブがキュー上に存在しているとき、ワーカは各ジョブ間にディレイを取らずに実行し続けます。sleepオプションは、新しく処理するジョブが存在しない時に、どの程度「スリープ」するかを秒単位で指定します。スリープ中、ワーカは新しいジョブを処理しません。ジョブはワーカが目を覚ました後に処理されます。

php artisan queue:work --sleep=3

Supervisor設定

Supervisorのインストール

SupervisorはLinuxオペレーティングシステムのプロセスモニタで、queue:workプロセスが落ちると自動的に起動します。UbuntuにSupervisorをインストールするには、次のコマンドを使ってください。

sudo apt-get install supervisor

Tip!! Supervisorの設定に圧倒されそうならば、Laravelプロジェクトのために、Supervisorを自動的にインストールし、設定するLaravel Forgeの使用を考慮してください。

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 sqs --sleep=3 --tries=3
autostart=true
autorestart=true
user=forge
numprocs=8
redirect_stderr=true
stdout_logfile=/home/forge/app.com/worker.log
stopwaitsecs=3600

この例のnumprocsディレクティブは、Supervisorに全部で８つのqueue:workプロセスを実行・監視し、落ちている時は自動的に再起動するよう指示しています。commandディレクティブのqueue:work sqsの部分を変更し、希望のキュー接続に合わせてください。

Note: 一番時間がかかるジョブが消費する秒数より大きな値をstopwaitsecsへ必ず指定してください。そうしないと、Supervisorは処理が終了する前に、そのジョブをキルしてしまうでしょう。

Supervisorの起動

設定ファイルができたら、Supervisorの設定を更新し起動するために以下のコマンドを実行してください。

sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start laravel-worker:*

Supervisorの詳細情報は、Supervisorドキュメントで確認してください。

失敗したジョブの処理

時より、キューされたジョブは失敗します。心配ありません。物事は計画通りに進まないものです。Laravelではジョブを再試行する最大回数を指定できます。この回数試行すると、そのジョブはfailed_jobsデータベーステーブルに挿入されます。failed_jobsテーブルのマイグレーションを生成するにはqueue:failed-tableコマンドを実行してください。

php artisan queue:failed-table

php artisan migrate

次にキューワーカの実行時、queue:workコマンドに--triesスイッチを付け、最大試行回数を指定します。--triesオプションに値を指定しないと、ジョブは１回のみ試行します。

php artisan queue:work redis --tries=3

さらに、--delayオプションを使用し、失敗してから再試行するまでに何秒待てばよいかをLaravelへ指定できます。デフォルトでは、時間を置かずに再試行します。

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

ジョブごとに失敗したジョブの再試行までの遅延を設定したい場合は、キュー投入するジョブクラスでretryAfterプロパティを定義してください。

/**
 * ジョブを再試行するまでに待つ秒数
 *
 * @var int
 */
public $retryAfter = 3;

リトライ時のディレイを決める複雑なロジックが必要になる場合は、キュージョブクラスで、retryAfterメソッドを定義してください。

/**
* ジョブを再取得する前に何秒待つか計算する
*
* @return int
*/
public function retryAfter()
{
    return 3;
}

ジョブ失敗後のクリーンアップ

失敗時にジョブ特定のクリーンアップを実行するため、ジョブクラスでfailedメソッドを直接定義できます。これはユーザーに警告を送ったり、ジョブの実行アクションを巻き戻すために最適な場所です。failedメソッドには、そのジョブを落とすことになったThrowable例外が渡されます。

<?php

namespace App\Jobs;

use App\AudioProcessor;
use App\Podcast;
use Throwable;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

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

    protected $podcast;

    /**
     * 新しいジョブインスタンスの生成
     *
     * @param  \App\Podcast  $podcast
     * @return void
     */
    public function __construct(Podcast $podcast)
    {
        $this->podcast = $podcast;
    }

    /**
     * ジョブの実行
     *
     * @param  \App\AudioProcessor  $processor
     * @return void
     */
    public function handle(AudioProcessor $processor)
    {
        // アップロード済みポッドキャストの処理…
    }

    /**
     * ジョブ失敗の処理
     *
     * @param  \Throwable  $exception
     * @return void
     */
    public function failed(Throwable $exception)
    {
        // 失敗の通知をユーザーへ送るなど…
    }
}

Note: failedメソッドは、ジョブがdispatchNowメソッドでディスパッチされた場合には呼び出されません。

ジョブ失敗イベント

ジョブが失敗した時に呼び出されるイベントを登録したい場合、Queue::failingメソッドが使えます。このイベントはメールやSlackにより、チームへ通知する良い機会になります。例として、Laravelに含まれているAppServiceProviderで、このイベントのコールバックを付け加えてみましょう。

<?php

namespace App\Providers;

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

class AppServiceProvider extends ServiceProvider
{
    /**
     * 全アプリケーションサービスの登録
     *
     * @return void
     */
    public function register()
    {
        //
    }

    /**
     * 全アプリケーションサービスの初期処理
     *
     * @return void
     */
    public function boot()
    {
        Queue::failing(function (JobFailed $event) {
            // $event->connectionName
            // $event->job
            // $event->exception
        });
    }
}

失敗したジョブの再試行

failed_jobsデータベーステーブルに挿入された、失敗したジョブを全部確認したい場合はqueue:failed Artisanコマンドを利用します。

php artisan queue:failed

queue:failedコマンドはジョブID、接続、キュー、失敗した時間、その他の情報をリスト表示します。失敗したジョブをジョブIDで指定すると、リトライ可能です。たとえば、IDが5の失敗したジョブを再試行するには、以下のコマンドを実行します。

php artisan queue:retry 5

必要に応じ、複数のIDやIDの範囲（数値IDを使用時）をコマンドへ指定できます。

php artisan queue:retry 5 6 7 8 9 10

php artisan queue:retry --range=5-10

失敗したジョブをすべて再試行するには、IDとしてallをqueue:retryコマンドへ指定し、実行してください。

php artisan queue:retry all

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

php artisan queue:forget 5

失敗したジョブを全部削除するには、queue:flushコマンドを使います。

php artisan queue:flush

不明なモデルの無視

Eloquentモデルをジョブで取り扱う場合は自動的にキューへ積む前にシリアライズし、ジョブを処理するときにリストアされます。しかし、ジョブがワーカにより処理されるのを待っている間にモデルが削除されると、そのジョブはModelNotFoundExceptionにより失敗します。

利便性のため、ジョブのdeleteWhenMissingModelsプロパティをtrueに指定すれば、モデルが見つからない場合自動的に削除できます。

/**
 * モデルが存在していない場合に、ジョブを削除する
 *
 * @var bool
 */
public $deleteWhenMissingModels = true;

ジョブイベント

Queueファサードにbeforeとafterメソッドを使い、キューされたジョブの実行前後に実行する、コールバックを指定できます。これらのコールバックはログを追加したり、ダッシュボードの状態を増加させたりするための機会を与えます。通常、これらのメソッドはサービスプロバイダから呼び出します。たとえば、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
{
    /**
     * 全アプリケーションサービスの登録
     *
     * @return void
     */
    public function register()
    {
        //
    }

    /**
     * 全アプリケーションサービスの初期処理
     *
     * @return void
     */
    public function boot()
    {
        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メソッドを使用し、ワーカがキューからジョブをフェッチする前に、指定したコールバックを実行できます。たとえば、直前の失敗したジョブの未処理のままのトランザクションをロールバックするクロージャを登録できます。

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

Laravel 7.x キュー