Laravel Horizonの接続まとめ。

始めに

今回業務で大量のCSVファイルをインポートする際に、
ジョブとキューを使用してインポート処理を行いました。
またその際にLaravel Horizonを利用してキューを管理していました。
初めて使用したため、セットアップなどに苦戦したため、
まとめておきたいと思います。

設計

今回使用している技術は以下になります。(一部)
laravel 9.44.0
PHP 8.1.8
Composer 2.4.4
MySQL 8.0.28

そもそもキューとは？

Laravelのキューシステムは、時間のかかるタスクをリクエストの処理から分離し、
非同期で実行するための方法を提供します。
これにより、ウェブアプリケーションの応答性を向上させることができます。
キューを利用する一般的な例としては、メール送信、大量データの処理、
ビデオのエンコードなどがあります。
キュー設定オプションでは、Amazon SQS、Redis、databaseなどが選べます。

キューの概念

キューは、実行すべきタスク（ジョブ）を順番に並べるためのものです。
ジョブは、後で実行するためにキューに追加され、キューワーカーによって順次処理されます。
このプロセスは、タスクの実行を非同期で行うことを可能にし、
アプリケーションのパフォーマンスとユーザーエクスペリエンスを向上させます。
ジョブは処理自体を担当するものになります。

Laravel Horizonとは？

Redisをキューとして使用することで、使用できるようになる管理ツールです。
ジョブがどのように処理されているか、
ちゃんと成功したのか失敗したのか結果を管理することで、
一元的にシステムがどうなっているかという情報を見れるように工夫されています。
便利ですがRedisキューでしか使用できません。

今回Laravel Horizonを使用した機能

今回は、管理画面から大量のマスターデータをインポートできる機能を実装する際に、
ジョブとキューを使用しました。
マスターは更新されるため、何度もインポートされる可能性があり、
一度目と二度目では、インポートするものが少し変わる想定でした。
これまではService層でCSVをインポートしていましたが、
その処理自体を、ジョブに登録して、非同期処理をさせるようにしてみました。
Laravel Horizonの画面でキューの状態を確認するまでの手順をまとめます。

DB設定を変更する

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

config/database.php

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

ジョブを作成して、ジョブに登録する

まずは一度目にマスターをインポートする時に使用するジョブを作成します。

php artisan make:job ProcessMasterImportJob

ジョブファイルに処理を記載します。

ProcessMasterImportJob

namespace App\Jobs;

use App\Factories\Interfaces\SplFileObjectFactoryInterface;
use App\Repositories\Interfaces\MasterRepositoryInterface;
use App\Services\Interfaces\AdminMasterServiceInterface;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;
use Illuminate\Validation\ValidationException;

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

    private string $csvName;

    /**
     * @param string $csvName
     */
    public function __construct(string $csvName)
    {
        $this->csvName = $csvName;
    }

    /**
     * @return void
     * @throws ValidationException
     */
    public function handle(): void
    {
　　　　　　　　　　　　　　　　// 依存性注入を__constructで直接行うと、問題が生じるため、handle内でappヘルパー関数を使用する
        $splFileObjectFactory = app(SplFileObjectFactoryInterface::class);
        $masterRepository = app(MasterRepositoryInterface::class);
        $adminMasterService = app(AdminMasterServiceInterface::class);

        Log::info('マスターCSVインポート処理開始');
        $csvData = $splFileObjectFactory->create(storage_path('app/admin_csv/'.$this->csvName));
        $csvData->setFlags(
            \SplFileObject::READ_CSV |
            \SplFileObject::READ_AHEAD |
            \SplFileObject::SKIP_EMPTY |
            \SplFileObject::DROP_NEW_LINE
        );
        if ($masterRepository->isMasterEmptyFromAdmin()) {
            $adminMasterService->doFirstTimeImport($csvData);
        } else {
            MasterDoLatestImportJob::dispatch($this->csvName)->onQueue('doLatestImport');
        }
        Log::info('マスターCSVインポート処理終了');
    }
}

次に二度目のインポートの際のインポート処理を行うジョブを作成します。

php artisan make:job MasterDoLatestImportJob

namespace App\Jobs;

use App\Factories\Interfaces\SplFileObjectFactoryInterface;
use App\Models\Master;
use App\Repositories\Interfaces\MasterRepositoryInterface;
use App\Services\Interfaces\AdminMasterServiceInterface;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\DB;
use Illuminate\Validation\ValidationException;

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

    private string $csvName;
    private static $CHANGE_TYPE_DEFAULT = '0';
    private static $CHANGE_TYPE_DELETE = '1';
    private static $CHANGE_TYPE_NEW = '3';
    private static $CHANGE_TYPE_UPDATE = '5';

    /**
     * @param string $csvName
     */
    public function __construct(string $csvName)
    {
        $this->csvName = $csvName;
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle(): void
    {
        $splFileObjectFactory = app(SplFileObjectFactoryInterface::class);
        $csvData = $splFileObjectFactory->create(storage_path('app/admin_csv/'.$this->csvName));
        $csvData->setFlags(
            \SplFileObject::READ_CSV |
            \SplFileObject::READ_AHEAD |
            \SplFileObject::SKIP_EMPTY |
            \SplFileObject::DROP_NEW_LINE
        );

        DB::transaction(function () use ($csvData) {
            $masterRepository = app(MasterRepositoryInterface::class);
            $adminMasterService = app(AdminMasterServiceInterface::class);

            foreach ($csvData as $key => $line) {
                // CSVから取得したデータを連想配列に変換
                $csvMaster = $adminMasterService->makeMasterArrFromCsvLine($line);
                // データのバリデーションを行う
                $masterValidator = Master::makeValidator($csvMaster);
                if ($masterValidator->fails()) {
                    throw new ValidationException($masterValidator);
                }
                switch ($csvMaster['change_type']) {
                    case self::$CHANGE_TYPE_UPDATE:
                        // データの更新処理
                        $masterRepository->updateFromAdmin($csvMaster);
                        break;
                    case self::$CHANGE_TYPE_DEFAULT:
                        break;
                    case self::$CHANGE_TYPE_NEW:
                        // データの新規作成処理
                        $masterRepository->createFromAdmin($csvMaster);
                        break;
                    case self::$CHANGE_TYPE_DELETE:
                        // データの廃止処理
                        $masterRepository->changeDisabledFromAdmin($csvMaster);
                        break;
                }
            }
        });
    }
}

ジョブファイルを用意したあとは、
コントローラーから最初のジョブファイルを呼び出すようにしてみます。

class MasterController extends Controller
{
  /**
     * マスターデータをCSVインポートする
     *
     * @param CsvRequest $request
     * @return CreatedResource
     * @throws \Exception
　　 */
  public function importAdminCsv(CsvRequest $request): CreatedResource
    {
        $logger = new ApplicationLogger(__METHOD__);
        $logger->write('[入力パラメータ]:' . print_r($request->all(), true));

        $logger->write('マスターCSVインポート機能処理開始');
        $this->masterService->importMasterFromAdmin($request->getMasterFilePath());
        $logger->success();
        try {
            $logger->write('マスターCSVインポート機能処理開始');
            $csvName = $request->file('sickness_csv_file')->getClientOriginalName();
            $request->file('admin_csv_file')->storeAs('admin_csv', $csvName, 'local');
            ProcessMasterImportJob::dispatch($csvName)->onQueue('import');
            $logger->success();
            return new CreatedResource();
        } catch (\Exception $e) {
            $logger->exception($e);
            throw $e;
        }
}

docker環境でRedisを使用している際の注意

これまで通り、以下の記述でCSVのファイルパスを取得しようとしたところ、
うまく取得できない問題が発生しました。

$csvData = $this->splFileObjectFactory->create($this->filePath);

そのためStorageにまずファイルを置くようにしました。

// なぜかlocalを指定しないと、minioが指定されたことになりエラーが出てしまう
$request->file('master_csv_file')->storeAs('admin_csv', $csvName,'local');
$csvData = $splFileObjectFactory->create(storage_path('app/admin_csv/'.$this->csvName));

ジョブファイルでは、置いてあるディレクトリからファイルパスを参照するようにすることで、
うまく読み取れるようになりました。

 $csvData = $splFileObjectFactory->create(storage_path('app/admin_csv/'.$this->csvName));

Laravel Horizonをインストールして、設定する

公式が一番わかりやすいので、こちらを参照してください。
https://readouble.com/laravel/8.x/ja/horizon.html
ただphp artisan horizonを行おうとして、下記のエラーが出ました。

root@bdcfb900eff6:/var/www# php artisan horizon
Horizon started successfully.

   Error 

  Call to undefined function Laravel\Horizon\Console\pcntl_async_signals()

  at vendor/laravel/horizon/src/Console/HorizonCommand.php:48
     44▕         ProvisioningPlan::get(MasterSupervisor::name())->deploy($environment);
     45▕ 
     46▕         $this->info('Horizon started successfully.');
     47▕ 
  ➜  48▕         pcntl_async_signals(true);
     49▕ 
     50▕         pcntl_signal(SIGINT, function () use ($master) {
     51▕             $this->line('Shutting down...');
     52▕ 

      +13 vendor frames 
  14  artisan:37
      Illuminate\Foundation\Console\Kernel::handle(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))

エラーで検索したところ、下記記事がヒット
https://stackoverflow.com/questions/54566977/laravel-horizon-throws-error-call-to-undefined-function-laravel-horizon-consol

原因は、ext-pcntlが必要だがインストールされていなかったことでした。
Dockerを使用しているため、Dockerfileに下記を記載しました。

RUN docker-php-ext-install pcntl

無事にLaravel Horizonの画面にアクセスできるようになりました。

失敗したジョブも見ることができ、どのコードの部分で失敗したかもわかります。
今は二つしかキューがないですが、
これが複数だった場合、どの処理が途中で止まったかや、
今どこまでジョブが進んでいるのかがわかりとても便利です。

まとめ

DockerでLaravel Horizonを使用しようとして、
意外に躓いた箇所が多かったため、
自分の備忘録としてまとめておきました。
誰かの参考になれば幸いです。

参考文献

この記事は下記の記事を参考にして、書かせていただきました。
https://qiita.com/KyuKyu/items/1316ab89703519427c56
https://eza-s.com/blog/archives/356/
https://readouble.com/laravel/8.x/ja/horizon.html
https://stackoverflow.com/questions/54566977/laravel-horizon-throws-error-call-to-undefined-function-laravel-horizon-consol