laravelのジョブ・キュー・ワーカーを試す。

laravel ジョブ・キューモデル

laravel には、ジョブ・ワーカーとキューがデフォルトで用意されています。

これを少し試してみました。

今回試すこと

キューにジョブをいれて、ワーカーで処理をする。

queue でできること

ジョブを待ち続けて実行する。

php artisan queue:work # 無限ループによるキュー待ち

キューに値が入ると処理を始めるワーカーを起動します。

今回やらないこと。（並列処理）

こんかいは並列処理試さない。特に複数マシンでジョブの共有。ジョブの並列起動。これらは試しません。ネットワーク超えてジョブを共有して処理をすれば便利なのだろうが考えることが多くなるので今回は試しません。

laravelのジョブキュー並列処理は、基本機能をあれこれ触って考えるよりも、Redis 使って処理するか、laravel/horizon を使うべきだと思う。

基本的に出来ないと思っておいたら無難ってだけで、実現する方法はいっぱいあしライブラリもいっぱいあるし、php8.1で進化したFiberとかもそのうちサポートされるだろうし、基本的なところを抑えておき基本的な機能で作っておくほうが最初は無難だと思われる。

laravel ジョブ・キューモデル-その１sync

ジョブの作成と実行

ジョブクラスを作成し、コマンドからジョブクラスを登録し、コマンドからジョブを実行する

ジョブクラスを作る app/Jobs/MyFirstJob.php が作成される。

php artisan make:job MyFirstJob

app/Jobs/MyFirstJob.php 編集してechoするだけの仕事を定義する。

<?php
class MyFirstJob implements ShouldQueue {
  use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
  
  public function handle () {
    echo "Hello from MyFirstJob\n";
  }
}

ジョブ登録する

ジョブを登録するために、artisan コマンドで登録専用のコマンドを作る。

php artisan make:command EnqueueMyFirstJob

app/Jobs/MyFirstJob.php ジョブ登録するコマンドを作る

<?php
class EnqueueMyFirstJob extends Command {
  protected $signature = 'myfirstjob:add';
  protected $description = 'MyFirstJobをキュー追加';
  
  public function handle () {
    MyFirstJob::dispatch();
    return 0;
  }
}

artisan を実行する(ジョブ登録する)

php artisan myfirstjob:add
Hello from MyFirstJob

ジョブはその場で実行されます。

即座に実行されます。Jobで定義した仕事は追加された瞬間に実行されます。

開発時やお仕事（ジョブ）の定義を作るときは、このままで大丈夫です。

しっかりジョブキューとして使うには、分離する必要があります。

laravel ジョブ・キューモデル-その２db

先程作ったジョブは、その場で実行されます。ジョブなのにその場で同期的に実行されました。これは、デフォルトの実行タイミングがsync となっていて、呼び出した側でそのまま実行されるからです。

次に、sync をやめて、ワーカーに処理を任せるように変えて、キューイングしたいと思います。

分離する

ジョブを即時処理せずに、データベースにためておく。

データベースにジョブをためておく事ができます。そのためにデータベースにテーブルを作ります。これは最初から作成キットがコマンドとして用意されていて、それを実行するだけです。

データベースにキュー管理テーブルを作る

php artisan queue:table
php artisan migrate

データベースにジョブを登録するコマンドを作ります。

artisan make:command EnqueueJobIntoDatabase

EnqueueMyFirstJobを改造します。

最初に作ったジョブ登録コマンドを改造して、データベースにキューとして積上げていきます。

<?php
class EnqueueJobIntoDatabase extends Command {
  protected $signature = 'myfirstjob:add_to_db';
  public function handle () {
    app('queue')->connection('database')->push(new MyFirstJob());
    return 0;
  }
}

ジョブを登録します。

php artisan myfirstjob:add_to_db

データベースに登録したジョブを実行します。

php artisan queue:work --once database

これで、登録と実行が分離されて、workder によりジョブが実行されます。

queue:work の実行書式について

実行書式がわかりにくいので、分割して見おきます。

artisan queue:work --once 接続名

登録されたジョブを１つだけ処理をする。

artisan queue:work --once

ジョブを取り出す接続先を指定する。

artisan queue:work 接続名

指定した接続先から１つ取り出して実行する。２オプションをまとめる

artisan queue:work --once 接続名

上記の実行方法が、最初の例で、ジョブを実行したコマンドになります。

実験中は、ジョブをまとめて処理する必要が殆どないので、１回ずつ --onece をつけて実行することがほとんどかなと思います。

ジョブの登録の箇所について

データベースにジョブを登録した箇所を見ておきます。

Laravelコンテナから、Queueクラスを取り出して実行している。

app('queue')->connection('database')->push(new MyFirstJob());

次と同じである。

$con = \Illuminate\Support\Facades\Queue::connection('database');
$con->push( new MyFirstJob() );

databaseとは、単純にデータベースを示す様に見えます。しかし実際には、laravel 設定から、databaseという名前で登録されたキューの名前から、接続を取り出しています。

ジョブの監視。（ワーカー実行）

データベースにジョブが登録されるまで、ずっと待ってキューが来たら処理をするのがワーカー

php artisan queue:listen  database

登録したら、ジョブが実行される。

画面左側が、ワーカーです。画面右側でジョブを登録をしています。

監視間隔

ワーカーは、無限ループで待ってくれる。実行間隔は１秒程度。

ちなみに、本番環境ではエラーを考慮して listenよりworkを使うかなと思います。

### デプロイ先（本番環境など）
php artisan queue:work  database

いい感じにフォークしてくれて、例外でも停止しないで監視を継続してくれます。（あとで試します）

ジョブ・キューの接続先の設定

毎回、接続先を指定してられないし、接続名がソースコードに残るのも好ましくない。

デフォルト接続先

そこで、接続先のデフォルトを変えておく

設定は、次のようになっている。デフォルトはsync でキューに入れずに直接実行です。

config/queue.php

 'default' => env('QUEUE_CONNECTION', 'sync'),

設定なので、環境変数env で指定するか、configで指定するか。の２択があります。

envで指定する場合

env ファイルに記述すればいいです。

QUEUE_CONNECTION=database

config で指定する場合。

設定ファイルに記述します。

config/queue.php

return [
 'default' => env('QUEUE_CONNECTION', 'datanase'),

...(snip)

configを変えたら、キャッシュのクリア。

artisan config:clear
artisan config:cache

これで、ひとまず、一通りのジョブキュー実行が行えるはずである。

キューの接続先を「設定ファイル」で候補を作っておき、本番用・実験用でデフォルトを切り替えておくのが便利なかと思います。

ジョブを１分間に１度まとめて処理する。

php artisan queue:work --sleep=60

６０秒に一度ジョブを全部取得して実行する。仕事がなくなったら sleep します。

このオプションから分かる通り、ワーカーは仕事をまとめて取りに行く。

60秒毎回チェックするというわけでなく、スリープ間隔が60秒ですね。

仮に、60秒ごとに起動したいなら schedule:work の方を使うべきでしょうね。

laravel ジョブ・キューモデル-その３クラス定義更新

worker に登録したら、ジョブのクラスがなくなったどうなるでしょうか。

ジョブはクラスを登録します。クラス定義を書いてジョブとして登録してるわけです、だったらキューに溜まったままクラス名を変えたらどうなるでしょうか。

キューに予約タイミングでロードなのか、実行にクラスをロードしているのか、laravelはどちらなのでしょうか。確認しておきます。

実行前にJobクラスを変更したら？

ジョブには、クラス名だけが登録されています。

キューの中身を確認する

ジョブの一覧を見るコマンドを作ってみます。

artisan make:command Joblist

クラスにジョブを吐き出させます。

<?php
class Joblist extends Command {
  protected $signature = 'queue:list';
  public function handle () {
    $ret = \DB::table('jobs')->get('payload');
    foreach( $ret as $idx=> $r){
      dump(json_decode($r->payload));
    }
    return 0;
  }
}

実行します。

 artisan queue:list

キューの中身はSerializeみたいです。

クラスがシリアライズされているのがわかります。

{#875
  +"uuid": "6f0c4b73-cde7-4da2-ae8e-6124a3622f23"
  +"displayName": "App\Jobs\MyFirstJob"
  +"job": "Illuminate\Queue\CallQueuedHandler@call"
  +"maxTries": null
  +"maxExceptions": null
  +"failOnTimeout": false
  +"backoff": null
  +"timeout": null
  +"retryUntil": null
  +"data": {#887
    +"commandName": "App\Jobs\MyFirstJob..."
    +"command": "O:19:"App\Jobs\MyFirstJob":10..."
  }

しかもクラス名もシリアライズされている。なのでオブジェクトをダンプしてるとはいえ、クラス定義まではダンプしていない事がわかります。

ジョブ（タスク）登録後にジョブのクラス定義を変えると、実行時タイミング最新版のコードで実行されるとわかります。登録時のクラスの定義までは持っていってないことがわかります。

ここから複数のワーカーを別マシンで動かすときに、クラスファイルがないと動かないこともわかります。つまりワーカー専用のマシンを用意し、複数のいろいろなlaravelプロジェクトからジョブを受け取って、複数のプロジェクトでワーカーをシェアする設計は面倒くさいとわかります。laravelのキュー・ワーカーを使っただけでは、aws lambdaみたいな関数実行専用サーバー作ることが困難であるとも言えます。

Jobクラス定義を変えてみます。

ジョブを登録します。

for i in {1..10}; do php artisan myfirstjob:add_to_db ; done

登録した段階では、まだワーカーを動かしません。

登録後、絶対失敗するジョブにししておきます。がまだ保存しません。

<?php
class MyFirstJob implements ShouldQueue {
  use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
   
  public function handle () {
    throw new \Exception('failed on purpose.');
    echo "Hello from MyFirstJob\n";
  }
}

ワーカーを起動したタイミングでクラス定義を更新しちゃいます。

起動と、コードの保存を併せてやります。

php artisan queue:listen  database --sleep=30

クラス定義変更で実行エラー

実行ログをみるとソースを保存した後。新しい定義にした箇所から、実行に失敗しログに出てきます。

php artisan queue:listen  database --sleep=30
[2021-08-23 10:15:11][41] Processed:  App\Jobs\MyFirstJob
[2021-08-23 10:15:12][42] Processing: App\Jobs\MyFirstJob
Hello from MyFirstJob
[2021-08-23 10:15:12][42] Processed:  App\Jobs\MyFirstJob
[2021-08-23 10:15:13][43] Processing: App\Jobs\MyFirstJob
Hello from MyFirstJob
[2021-08-23 10:15:13][43] Processed:  App\Jobs\MyFirstJob
[2021-08-23 10:15:14][44] Processing: App\Jobs\MyFirstJob
[2021-08-23 10:15:14][44] Failed:     App\Jobs\MyFirstJob
[2021-08-23 10:15:15][45] Processing: App\Jobs\MyFirstJob
[2021-08-23 10:15:16][45] Failed:     App\Jobs\MyFirstJob
[2021-08-23 10:15:17][46] Processing: App\Jobs\MyFirstJob
[2021-08-23 10:15:17][46] Failed:     App\Jobs\MyFirstJob

ということで、ジョブ定義をキューに溜まってるときは変えるべきじゃありません。

キューイングされているジョブ定義のクラス変更が困難です。

このことから、git push からの自動デプロイに懸念が残ります。

本番環境で git pull してソース更新するとき、一旦ワーカーを止めておかないと誤作動しそうですね。

今回は、キュー先がデータベースで試しましたが。Redisでも多分同じだろうと思います。試してません。

2024-01-31

fix typo

それマグで！

知識はカップより、マグでゆっくり頂きます。　takuya_1stのブログ

laravelのジョブ・キュー・ワーカーを試す。

今回試すこと

queue でできること

今回やらないこと。（並列処理）

laravel ジョブ・キューモデル-その１sync

ジョブの作成と実行

ジョブはその場で実行されます。

laravel ジョブ・キューモデル-その２db

分離する

queue:work の実行書式について

ジョブの登録の箇所について

ジョブの監視。（ワーカー実行）

監視間隔

ジョブ・キューの接続先の設定

デフォルト接続先

config/queue.php

envで指定する場合

config で指定する場合。

config/queue.php

ジョブを１分間に１度まとめて処理する。

laravel ジョブ・キューモデル-その３クラス定義更新

実行前にJobクラスを変更したら？

キューの中身を確認する

キューの中身はSerializeみたいです。

Jobクラス定義を変えてみます。

クラス定義変更で実行エラー

2024-01-31