LaravelでJobとQueueを触ってみる

はじめに

1年前にJobを使用する際、動かしながら残していた実装メモを投下させていただきます。
情報が古かったり、間違いがあればご指摘いただければ幸いです。

使用言語:Laravel(8.X)

①jobをスタックする為のテーブルを作成する。

【コマンド】

php artisan queue:table

• ジョブをキューに保持するテーブルを作成する為のmigrationファイルを作成。
• [App\Jobs;]　配下にjobのClassも一緒に作成される。
  ※実行順序は、先入れ先出し方。

【コマンド】

[php artisan queue:failed-table]

• 失敗ジョブを登録するテーブルを作成する為のmigrationファイルを作成。

【作成されるmigrationファイル】

<?php
use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;
return new class extends Migration
{
    /**
     * Run the migrations.
     *
     * @return void
     */
    public function up()
    {
	   //下記が自動的に生成される。
        Schema::create('jobs', function (Blueprint $table) {
            $table->bigIncrements('id'); //ユニークid
            $table->string('queue')->index();  //INERT時、「default」が入る、ジョブが属するキューの名前
            $table->longText('payload'); //ジョブ実行内容を保持するカラム→キューとも呼ぶ：JSON形式で保持
            $table->unsignedTinyInteger('attempts'); //エラー試行回数をカウント
            $table->unsignedInteger('reserved_at')->nullable(); //ジョブが処理のために「予約」または「ロック」された日時を記録。これは、ジョブがキューから取り出され、ワーカーによって処理が開始された時刻を示す。
            $table->unsignedInteger('available_at'); //ジョブが処理可能になる日時を記録。これは、ジョブがキューに追加された日時、または遅延ジョブの場合には実行可能になる未来の日時を示す。
            $table->unsignedInteger('created_at'); //作成日
        });
    }
    /**
     * Reverse the migrations.
     *
     * @return void
     */
    public function down()
    {
        Schema::dropIfExists('jobs');
    }
};

EX【作成されるレコードのイメージ】

Jobキック時失敗したキューは、failed_jobsテーブルへINSERTされる。
　→exceptionカラムにはエラーログに記述される例外の記述内容と同じ内容が格納される。
　※下記は例として重複するメアドでuserレコードをINSERTしようとして失敗させたもの。

②jobのロジックを記述するClassを実装

【サンプルソース】

<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use App\Models\User;
class TestJobs implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    /*グローバル変数で$triesを指定すると試行回数を指定できる。
     *「1」なら二度目の試行無し, 「2」ならエラー後一回試行, 「0」は無限に試行する為注意
     * 試行する度、jobsテーブルの「attempt」カラムへ試行回数を更新
　　　*　　→下記で指定した回数(3ならエラー後2回試行)を超えた場合は、jobsテーブルからはレコードが削除され失敗を記録するテーブルへ回る
　　　*/
    public $tries = 3;


    private $test_data;
    /**
     * Create a new job instance.
     * (呼び出し元でジョブをスタック「dispatch」メソッドを実行する時に引数に渡した値がここへ渡る)
     *
     * @return void
     */
    public function __construct($val)
    {
        //データセットする。
        $this->test_data = $val;
    }
    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        //ここで実行した内容は、jobsテーブルの「payload」カラムでオブジェクト形式で保持される。
        //またhandleメソッドが走るとjobsテーブルへINSERTされる。
        //注意：この段階ではジョブをキューにスタックするだけでジョブが実行される訳ではない。
        //また、ジョブが実行されるとjobsテーブル内にスタックしたキュー(レコード)はDELETEされる。
        //ジョブをキックした結果、成功しても失敗してもキューは削除される
        $test = new User;
        $test->name = $this->test_data["name"];
        $test->email = $this->test_data["email"];
        $test->password = $this->test_data["password"];
        $test->created_at = now();
        $test->updated_at = now();
        $test->save();  //ユーザーを作成。
        
        /**
		 *現場で使用する時の想定
		 *
         * CSVアップロードで該当テーブルへINSERT
         * INSERTされたレコードからマクロ実行し、値を整形しUPDATE (ここでJobをキック)
         * 整形後の値をIF間でデータ連携させる為、CSV出力→WIndowsサーバへアップロード
         * 別プロジェクトからアップロードされたCSVをDLし連携完了の流れ
         */
    }
}

③jobをdispatchして、キューへスタックする。

※dispatchする場所は、キューへスタックする任意のトリガーで実行。
※あくまでキューへスタックしているだけで、ここの段階ではまだキューはキックされない。

//dispatch()でジョブをスタック←use したTestJobs Classのhandle()内の処理が走る。
//dispatch()の引数に値を渡すと、TestJobsのコンストラクタに値が渡せる。
$test_data = [
    "name" => "job_name",
    "email" => "job_email",
    "password" => "job_password",
];
        
TestJobs::dispatch($test_data);

④スタックしたキューをキックする

私が経験した現場ではWEBジョブで実行したりしていました。

【local環境で動かしてみる】

php artisan queue:work　

→実行コマンドを実行すると監視モードに入る
→ctrl + cで抜け出すまでは、ジョブがスタックされる度に自動実行される。

php artisan queue:work --once

→一回実行

おまけ

EX.JobClassで指定できるプロパティ一覧

1. $delay: ジョブが処理されるまでの遅延時間を指定。
2. $tries: ジョブが失敗した場合に再試行する最大回数を指定。
3. $maxExceptions: ジョブが失敗と見なされる前に、許容される最大例外数を指定。
4. $timeout: ジョブがタイムアウトになるまでの最大秒数を指定。
   　　　　 指定した時間を超過した場合、ジョブは失敗とみなす。
5. $retryUntil: ジョブが再試行される最終時刻を指定。
6. $connection: ジョブが処理されるべきキューコネクションを指定。
7. $queue: ジョブが送信されるキューの名前を指定。
8. $afterCommit: このプロパティが「true」に設定されている場合、ジョブはデータベースのトランザクションがコミットされた後にのみディスパッチされる。

※上記のプロパティを使用する時は下記の様な形で指定する。
　地味にプロパティの設定で詰まった記憶があるのでご参考いただけると...

.
.
.

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

    /**
     * ジョブが失敗した場合に再試行する最大回数を指定。
     *
     * @var int
     */
    public $tries = 3;

    /**
     * ジョブがタイムアウトになるまでの最大秒数を指定。
     *
     * @var int
     */
    public $timeout = 60;

.
.
.

EX.Workerコマンド一覧

1.キューワーカーを起動し、ジョブを処理する。
このコマンドにオプションを指定する事で下記の様な事ができる。
(例：--queue で処理するキューを指定、--delay で失敗したジョブの再試行までの遅延を設定、--memory でワーカーのメモリ制限を設定など)

php artisan queue:work

2.queue:workコマンドと同様にジョブを処理するが、各ジョブの後に新しいプロセスを起動させる。コードの変更を即座に反映できるが、パフォーマンスは低下する。

php artisan queue:listen

3.全てのキューワーカーをリスタート。

php artisan queue:restart

4.失敗したジョブを再試行。特定のジョブIDを指定 or allオプションを使用して全ての失敗したジョブを再試行できる。

php artisan queue:retry

5.失敗したジョブを失敗したジョブのリストから削除。

php artisan queue:forget

6.全ての失敗したジョブを失敗したジョブのリストから削除。

php artisan queue:flush

7.失敗したジョブのリストを表示。

php artisan queue:failed

8.失敗したジョブを保存するためのデータベーステーブルを作成します。

php artisan queue:failed-table

9.古い失敗したジョブをデータベースから削除。

php artisan queue:prune-failed