LaravelのJOBとQUEUEを使用して非同期処理を実装してみた。

こんにちは。遠藤です。

今回はLaravelのJOBとQUEUEを使用して非同期処理を実装してみたので、つまずいた点にも触れながら実装方法について解説させていただきます。

ちなみに今回実装した非同期処理は以下のようなものです。

A：新規会員入会 → B：会員登録処理実施（ → C：非同期で関連する会員のDB情報更新） → 処理終了

Cの処理が重い処理だったため、非同期での実装を試みました。

JOBとQUEUEについて
.envの設定
DBにQUEUE用のテーブルを作成
JOBクラスの作成
QUEUEへ格納
WorkerによるJOBの実行
SupervisorによるWorkerの管理

JOBとQUEUEについて

LaravelにおいてJOBとは処理そのもののことです。例えば、会員管理システムにおいて会員のDB情報を更新するする処理のことをJOBと呼びます。

LaravelにおいてQUEUEとはJOBをため込む箱のようなものです。最初に箱に入ったJOBから順番に処理していく仕組みです。このQUEUEを使用することでJOBを非同期で実行することができます。QUEUEはデータベース等を用いて用意します。今回はデータベースをQUEUEとして使用したのでその方法を紹介します。

.envの設定

.envファイルのQUEUE_CONNECTION変数の設定を以下に変更します。

QUEUE_CONNECTION=database

1	QUEUE_CONNECTION=database

最初この設定を忘れており、デフォルトの以下の設定になっていました。

QUEUE_CONNECTION=sync

1	QUEUE_CONNECTION=sync

syncになっているとデータベースに格納されません。そのため非同期実行ではなく同期実行となってしまいます。

また、.envファイルのQUEUE_DRIVER変数の設定も追加します。

QUEUE_DRIVER=database

1	QUEUE_DRIVER=database

以上で.envの設定は終了です。

※ データベース接続設定はできている前提です。

DBにQUEUE用のテーブルを作成

JOBをため込むためのテーブルを作成します。

テーブルは以下のartisanコマンドを使用することで自動で作成できます。

php artisan queue:table
php artisan migrate

1 2	php artisan queue:table php artisan migrate

以下はマイグレーションファイルの中身です。

<?php 
use Illuminate\Database\Migrations\Migration; 
use Illuminate\Database\Schema\Blueprint; 
use Illuminate\Support\Facades\Schema; 

class CreateJobsTable extends Migration { 
/**
 * Run the migrations. 
 * 
 * @return void
 */ 
  public function up() { 
    Schema::create('jobs', function (Blueprint $table) { 
      $table->bigIncrements('id');
      $table->string('queue')->index();
      $table->longText('payload');
      $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');
  }
}

<?php

use Illuminate\Database\Migrations\Migration;

use Illuminate\Database\Schema\Blueprint;

use Illuminate\Support\Facades\Schema;

class CreateJobsTable extends Migration {

/**

* Run the migrations.

* @return void

public function up() {

Schema::create('jobs', function (Blueprint $table) {

$table->bigIncrements('id');

$table->string('queue')->index();

$table->longText('payload');

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

}

migrationを使っていないという方は以下のSQLをご使用ください。

CREATE TABLE `jobs` (
`id` bigint unsigned NOT NULL AUTO_INCREMENT,
`queue` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL,
`payload` longtext COLLATE utf8mb4_unicode_ci NOT NULL,
`attempts` tinyint unsigned NOT NULL,
`reserved_at` int unsigned DEFAULT NULL,
`available_at` int unsigned NOT NULL,
`created_at` int unsigned NOT NULL,
PRIMARY KEY (`id`),
KEY `jobs_queue_index` (`queue`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

CREATE TABLE `jobs` (

`id` bigint unsigned NOT NULL AUTO_INCREMENT,

`queue` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL,

`payload` longtext COLLATE utf8mb4_unicode_ci NOT NULL,

`attempts` tinyint unsigned NOT NULL,

`reserved_at` int unsigned DEFAULT NULL,

`available_at` int unsigned NOT NULL,

`created_at` int unsigned NOT NULL,

PRIMARY KEY (`id`),

KEY `jobs_queue_index` (`queue`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

JOBクラスの作成

JOBクラスをapp/jobsに作成したます。

以下のartisanコマンドで自動作成できます。

php artisan make:job UpdateMemberInfoJob

以下app/jobs/UpdateMemberInfoJob.phpの中身です。

handleの中に処理を記述していきます。

<?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\Http\Controllers\MemberController; 

class UpdateMemberInfoJob implements ShouldQueue { 

  use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; 
  /**
   * Create a new job instance.
   *
   * @return void
   */ 
  public function __construct() { 
    //
  } 

  /**
   * Execute the job.
   * 
   * @return void
   */ 
  public function handle() { 
    
    // 非同期で実行する処理（JOB）を記述
    $memberController = new MemberController; 
    $memberController->updateMemberInfo();
  }
}

<?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\Http\Controllers\MemberController;

class UpdateMemberInfoJob implements ShouldQueue {

use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

/**

* Create a new job instance.

* @return void

public function __construct() {

}

/**

* Execute the job.

* @return void

public function handle() {

// 非同期で実行する処理（JOB）を記述

$memberController = new MemberController;

$memberController->updateMemberInfo();

}

QUEUEへ格納

JOBクラスができたら、dispatchメソッドを呼び出すことで、QUEUEへJOBを格納します。

JOBは非同期実行されるため、処理Aの後、すぐに処理Bが実行されます。

<?php

namespace App\Http\Controllers;

use Illuminate\Http\Request;
// JOBを実行したい箇所に追記する
use App\jobs\UpdateMemberInfoJob;

class EntryController extends Controller
{
    public function entry(Request $request)
    {
        // 処理A（省略）
        // JOBをQUEUEに格納
        UpdateMemberInfoJob::dispatch();
        
        // 処理B（省略）
    }
}

<?php

namespace App\Http\Controllers;

use Illuminate\Http\Request;

// JOBを実行したい箇所に追記する

use App\jobs\UpdateMemberInfoJob;

class EntryController extends Controller

{

public function entry(Request $request)

{

// 処理A（省略）

// JOBをQUEUEに格納

UpdateMemberInfoJob::dispatch();

// 処理B（省略）

}

WorkerによるJOBの実行

上記でQUEUEにため込んだJOBを実行するのがWokerです。

entryメソッドが実行されるたびに、QUEUE（jobsテーブル）にJOBがたまっていきます。

以下のコマンドにより、Workerを起動させることでJOBが実行されます。

control + – 等で抜けるまで順次JOBを処理し続けます。

php artisan queue:work

1 2	php artisan queue:work

SupervisorによるWorkerの管理

QUEUEにたまったJOBをすぐに処理したい。と思っていましたが、上記の方法でのWorker起動ですとリリース時にコマンドを実行しておいたもののWEBサーバーの再起動等の何らかの理由でWorkerが終了してしまった場合、QUEUEにたまったJOBが処理されないという問題が発生してしまいます。そこでlinuxのSupervisorを使用することでコマンドを管理することにしました。

// Supervisorインストール
rootログイン
yum -y install supervisor
systemctl start supervisord
systemctl enable supervisord
systemctl status supervisord

// Supervisorにworker設定
vi /etc/supervisord.conf
以下を設置
  ```
  [program:laravel-worker]
  process_name=%(program_name)s_%(process_num)02d
  command=php /var/www/app_name/artisan queue:work
  autostart=true
  autorestart=true
  user=root
  numprocs=1
  startsecs=0
  redirect_stderr=true
  stdout_logfile=/var/log/queue_worker/worker.log
  stopwaitsecs=3600
 ```

// ログ出力ディレクトリ作成
mkdir /var/log/queue_worker

// 再起動
systemctl restart supervisord
systemctl status supervisord

// Supervisorインストール

rootログイン

yum -y install supervisor

systemctl start supervisord

systemctl enable supervisord

systemctl status supervisord

// Supervisorにworker設定

vi /etc/supervisord.conf

以下を設置

```

[program:laravel-worker]

process_name=%(program_name)s_%(process_num)02d

command=php /var/www/app_name/artisan queue:work

autostart=true

autorestart=true

user=root

numprocs=1

startsecs=0

redirect_stderr=true

stdout_logfile=/var/log/queue_worker/worker.log

stopwaitsecs=3600

```

// ログ出力ディレクトリ作成

mkdir /var/log/queue_worker

// 再起動

systemctl restart supervisord

systemctl status supervisord

numprocsを1に設定することで1つずつ処理が実行されるようになります。

JOBとQUEUEについて

.envの設定

DBにQUEUE用のテーブルを作成

JOBクラスの作成

QUEUEへ格納

WorkerによるJOBの実行

SupervisorによるWorkerの管理

❏❏ TOPIC ❏❏ ------------------------------------------------------------

カスタム自由!フリーECサイトパッケージ

チャットボット導入サービス

WEBシステム開発・スマホアプリ開発はSRIAへ