Symfony Advent Calendar 2018 11日目の記事です。

SymfonyでWebアプリケーションを作るとき、HTTPリクエストから直接実行するには重い処理があったらどうするか？
処理をコマンド化して、バックグラウンドで（非同期）実行させることが多いと思います。

Symfonyアプリケーションからコマンドを非同期で実行する仕組みとして、Symfony2時代には JMSJobQueueBundle が事実上の標準となっていました。
JMSJobQueueBundleは、Webアプリケーションの通常の処理の中からデータベース上の所定のテーブルにコマンド名や引数・オプション等を保存（予約）したものを、順に実行していってくれるバンドルです。
専用のコマンド jms-job-queue:run をsupervisordを使ってバックグラウンドで常駐させておくことで、予約したコマンドが次々に実行されていきます。

しかし長らくSymfony4に対応されず、代替方法を探す必要がありました。
今回はそのひとつ、 php-enqueue の async-command 機能について使い方をご紹介します。

※ ちなみについ最近（2018年11月）、ようやくJMSJobQueueBundleが Symfony4に対応しました!
内部の実装では結構変更があったのですが、動作のカスタマイズなしでJMSJobQueueBundleを単に使っている分には特に変更なく使えました。 JMSJobQueueBundleは広く使われているので、今後Symfony4時代になっても標準でありつづけることになりそうですね。

php-enqueueのasync-commandをジョブキューとして使う

enqueue/enqueue(php-enqueue)とは

FormaProという海外の開発会社が開発しているPHPのキュー実装です。JavaのJMSというキュー実装を参考に作られています。
統一的なインターフェイスで様々なキューにメッセージを送ることができます。

例えば、SQSキューにメッセージを送りたい場合、生のままのAWSのSDKを使って送信機能を開発すると、バックエンドのキューを変えることになったときにアプリケーション側で変更が多くなってしまいます。php-enqueueを使うと、バックエンドのキューを変えてもアプリケーション側の変更点は設定ぐらいになります（バックエンドのRDBMSが何であってもアプリケーション層からはDoctrineを使う、みたいな感じですね）

Symfonyから使う場合は enqueue/enqueue-bundle を使うと便利です。
といっても、 EnqueueBundle はただのキュー送受信の仕組みなので、キューにメッセージを送ったりキューからメッセージを読み出したりするだけで、JMSJobQueueBundleのように非同期でSymfonyコマンドを実行する機能まではついていません。

enqueue/async-commandとは

enqueue/async-command は、php-enqueueをキューとして使って非同期でSymfonyコマンドを実行するライブラリです。
JMSJobQueueBundleの代替として検討するため、早速使ってみましょう。

インストール

composerで enqueue/async-command と関連パッケージを依存に追加します。

$ composer require enqueue/async-command:"^0.9" enqueue/fs enqueue/enqueue-bundle symfony/process

composerからレシピを使うか尋ねられるのでyesを選ぶと Enqueue\Bundle\EnqueueBundle が自動で有効化されます。
enqueueの設定で、async-commandsを使う設定をします。

# config/packages/enqueue.yaml
enqueue:
  default:
    transport: '%env(ENQUEUE_DSN)%'
    client: ~
+    async_commands: true

今回キューのバックエンドとしてローカルファイルシステム enqueue/fs を使うので、キューに送られたメッセージを保存するためのファイルパスを ENQUEUE_DSN として設定します。

# .env
###> enqueue/fs ###
-ENQUEUE_DSN=file:///var/enqueue
+ENQUEUE_DSN=file:///path/to/project/async-command-demo/var/enqueue
###< enqueue/fs ###

コマンド実行をキューに入れる

ProducerInterface::sendCommand() で実行したいコマンドをメッセージ化してキューに送ることができます。

<?php

use Enqueue\Client\ProducerInterface;
use Enqueue\AsyncCommand\Commands;
use Enqueue\AsyncCommand\RunCommand;
use Symfony\Component\DependencyInjection\ContainerInterface;

/** @var $container ContainerInterface */

/** @var ProducerInterface $producer */
$producer = $container->get(ProducerInterface::class);

$producer->sendCommand(Commands::RUN_COMMAND, new RunCommand('some:command', ['arg1', 'arg2'], ['--foo' => 'bar']));

5分後のコマンド実行をキューに入れる

ProducerInterface::sendCommand() で送るメッセージを Enqueue\Client\Message でラップして、delayを指定することでコマンド実行を指定時間になるまで遅らせることができます。

<?php

use Enqueue\Client\Message;
use Enqueue\Client\ProducerInterface;
use Enqueue\AsyncCommand\Commands;
use Enqueue\AsyncCommand\RunCommand;
use Symfony\Component\DependencyInjection\ContainerInterface;

/** @var $container ContainerInterface */

/** @var ProducerInterface $producer */
$producer = $container->get(ProducerInterface::class);

$job = new Message(new RunCommand('some:command', ['arg1', 'arg2'], ['--foo' => 'bar']));
$job->setDelay(300 * 1000); // ドキュメントでは秒をセットすることになっているがなぜかミリ秒？
$producer->sendCommand(Commands::RUN_COMMAND, $job);

ジョブを実行するコマンド

キューを読んで処理するコマンドを起動すると、予約したコマンドが実行されます。

$ ./bin/console enqueue:consume --setup-broker

※ 2018年12月12日現在の最新バージョン enqueque/enqueue:0.9.0 ではキューに入れるメッセージのID保存の仕組みに バグ があり、動かない状態となってしまいました :sweat_drops:

残る課題

JMSJobQueueBundleからの移行先として見た場合、下記2点の課題がありました。

  • キューに入れ終わった際にジョブIDを返す仕組みがない
  • ジョブ実行が終わるとレコードがenqueueテーブルから削除されてしまう（キューのバックエンドとして enqueue/dbal を使った場合でも）

JMSJobQueueBundleではジョブのIDが返され、ジョブ完了（成功でも失敗でも）後もジョブのレコードがデータベースに保存され続けていましたが、php-enqueueのasync-commandにはその機能がありません。
async-commandを使うとしたら、ジョブの実行ログを保存したい場合はメッセージの返信をログする仕組みを、ジョブが成功したかどうかを追跡したい場合にはジョブステータス管理用のエンティティを自前で実装してステータス管理する必要があります。

まとめ

結果的にJMSJobQueueBundleの便利な点を再確認することになりました。
メジャーなバンドルが次々とSymfony4対応を完了させて、いよいよSymfony4時代の到来ですね。

参考リンク