Home > AWS | Laravel | PHP > Laravel の Queue で非同期処理を実装する(beanstalkd / IronMQ / SQS)

Laravel の Queue で非同期処理を実装する(beanstalkd / IronMQ / SQS)

この記事の所要時間: 1654

Laravel で実装されている Queue について見てみました。

laravel

Laravel では Queue を使うことで、時間がかかる処理や、時間差で実行したい処理を非同期で実行することができます。

Laravel 4.2 の Queue では、以下の 5 つのキュードライバをサポートしています。

  • sync
  • Beanstalkd
  • Amazon SQS
  • IronMQ
  • Redis

ここでは、sync、Beanstalkd、IronMQ、Amazon SQS について試してみました。

Laravel での設定

Laravel で Queue を使うには、app/config/queue.phpにて、利用するキューエンジンの選択、設定を行います。

もちろん他の設定と同じく、app/config/local/queue.phpapp/config/production/queue.phpなど、環境に応じて設定を切り替えることも可能です。

下記が設定例です。defaultでは利用するキュードライバを指定します。指定できる値は、sync, beanstalkd, sqs, iron, redis の 5 種類です。

connectionsキーでは、それぞれのドライバについて接続情報を指定します。下記では、beanstalkdの接続情報を指定しています。

return [
    'default' => 'beanstalkd',
    'connections' => [
        'beanstalkd' => [
            'driver' => 'beanstalkd',
            'host' => 'localhost',
            'queue' => 'default',
            'ttr' => 60,
        ],
    ],
];

キューへジョブを登録

キューへのジョブを登録するには、Queueクラスのpushメソッドを使います。

第一引数にはジョブを実行するワーカーのクラス名を、第二引数にはワーカーに渡すパラメータを指定します。

use CarbonCarbon;

Queue::push('MyWorker', ['message' => 'Hello!', 'date' => Carbon::now()]);

ここでは、ワーカーのクラス名のみを指定しているため、ジョブを実行する際はfireメソッドが呼ばれます。他にもMyWorker@doJobとすることで任意のメソッドを実行したり、クロージャを渡して、そのクロージャをワーカーとして実行したりできます。

詳しくはドキュメントを参照してください。

http://laravel.com/docs/queues
http://laravel4.kore1server.com/docs/42/queues

ワーカーの実装

ジョブを処理するワーカーとしてMyWorkerクラスを実装します。ワーカーは特定のクラスを継承する必要はなく、 POPO(Plain Old PHP Object)で良いです。

MyWorkerクラスにはfireメソッドを実装します。ジョブがキューに登録されると、リスナーからこのメソッドが呼ばれます。

第一引数にはジョブインスタンス、第二引数では、Queue::push()の第二引数で指定したパラメータが渡されます。ただ、このパラメータはジョブに入る時にjson_encode()されるので、オブジェクトのインスタンスはプロパティの連想配列となります。

下記の実装では、渡されたパラメータを元に文字列を生成して、echoで標準出力へ出力しているだけです。

ジョブが正常に完了したら、delete()メソッドで、ジョブを削除しておきます。これを行わないと同じジョブが何度も実行されています。

<?php
use CarbonCarbon;
use IlluminateQueueJobsJob;

class MyWorker
{
    /**
     * @param Job $job
     * @param array $data
     */
    public function fire(Job $job, array $data)
    {
        echo sprintf('[%s] %s at %s', Carbon::now(), $data['message'], $data['date']['date']) . PHP_EOL;
        $job->delete();
    }
}

キューの監視

キューを監視してワーカーを起動するリスナーを起動します。

Laravel では artisan コマンドですでに用意されているのでこれを利用します。

php artisan queue:listenコマンドを実行すると、キューを監視状態になります。この状態でキューにジョブが登録されていると、自動でワーカーが起動して処理が実行されます。

$ php artisan queue:listen

この状態でジョブが登録されると下記のように出力されます。

$ php artisan queue:listen
[2014-08-14 16:52:35] Hello! at 2014-08-14 16:52:34.000000
Processed: MyWorker

このプロセスが終了しているとキューにジョブが登録されても実行されないので、supervisor や monit などで常時起動するように設定しておくと良いでしょう。

一定時間後に実行するジョブ

Queue::push()で登録したジョブは、リスナーが感知するとワーカーが起動して実行されます。

それとは別に、Queue::later()というメソッドを使うと、一定時間経過後に実行するジョブを登録することができます。

第一引数には、ジョブの実行を開始する時間を指定します。数値を渡すと秒数として認識され、その秒数が経過した際にジョブが実行されます。Carbonクラスのインスタンスを渡すとCarbonクラスで指定された日時に実行されるジョブとして登録されます。

第二引数と第三引数は、Queue::push()の第一引数と第二引数と同じです。

下記の例では、10秒後に実行されるジョブを登録しています。

Queue::later(10, 'MyWorker', ['message' => 'Delayed', 'date' => Carbon::now()]);

リスナー側では下記のような出力になります。[]内の日時と、atの後ろの日時で 10 秒ずれていることが分かります。

$ php artisan queue:listen
[2014-08-14 16:52:45] Delayed at 2014-08-14 16:52:34.000000
Processed: MyWorker

これはローカルの beanstalkd での実行なのでずれが無いですが、IronMQ を利用した際は、数秒のずれがありました。多少のずれは発生するので、それを認識した上で利用すると良いでしょう。

サポートしているキュードライバ

Laravel でサポートしているキュードライバ(Redisを除く)を見てみましょう。

sync

デフォルトで指定されているドライバです。

仕組みとしてはキュー、ワーカーの流れを通るのですが、ジョブが登録されると、即時にワーカーが実行されます。

非同期処理にはならないので、実際にキューを使う場面では、他のドライバを利用する必要があります。

Beanstalkd

Beanstalkd は、オープンソースのキューイングシステムです。

利用するには、Beanstalkd 自体のインストールが必要になります。

Beanstalkd は、yum などパッケージでも公開されているので、利用するプラットフォームごとに選択してインストールすると良いです。

ここでは OSX 環境を想定して、Homebrew でインストールします。

$ brew install beanstalkd

beanstalkd を起動します。デフォルトではポート11300で待ち受けます。

$ beanstalkd

なお、beanstalkd は、デフォルトではキューの情報を永続化しません。このままだと beanstalkd が落ちるとキューの情報が消えてしまうので、本番環境などで運用する際は、永続化するように設定を行うのが良いです。

http://kr.github.io/beanstalkd/

Laravel で利用する際は、composer で下記のパッケージを追加しておきます。pda/pheanstalk は、3.x がリリースされていますが、Laravel 4.2 は、2.x 系にしか対応していないので、バージョン番号に注意して下さい。

"require": {
  "pda/pheanstalk": "~2.1.0"
}

IronMQ

Iron.io が提供しているメッセージキューサービスです。

ブラウザから登録するだけで利用することができます。基本は有償サービスなのですが、無料プランが用意されており、1,000,000APIリクエスト/月まで利用することができます。

グラフィカルな管理画面が用意されており、キューの状況などが分かりやすいのが良いです。

Heroku の Addon としても提供されているので、Heroku へデプロイするなら簡単に連携することができます。

Laravel で利用する際は、composer で下記のパッケージを追加しておきます。

"require": {
  "iron-io/iron_mq": "~1.5.1"
}

下記が Heroku 環境で IronMQ を利用する際の設定例です。Heroku では、接続情報が環境変数で提供されてるので、それらをtokenprojectに設定しています。

return [
    'default' => 'iron',
    'connections' => [
        'iron' => [
            'driver' => 'iron',
            'host' => 'mq-aws-us-east-1.iron.io',
            'token' =>  getenv('IRON_MQ_TOKEN'),
            'project' => getenv('IRON_MQ_PROJECT_ID'),
            'queue' => 'sample',
            'encrypt' => true,
        ],
    ],
];

AWS US-EastAWS EU-WestRackspace ORDRackspace LONのインスタンスを利用することができます。日本国内のインスタンスは存在しないので、国内のアプリケーションからキューを登録する場合はレイテンシが気になるかもしれません。

http://www.iron.io/

SQS

AWS のメッセージキューサービスです。

こちらも管理画面からキューを作成するだけで利用することができます。有償サービスですが、無料枠が用意されており、100万件キューイングリクエスト/月まで無料で利用することができます。

東京リージョンを利用できるので、アプリケーションサーバが国内にあるなら利用しやすいですね。

Laravel で利用する際は、composer で下記のパッケージを追加しておきます。

"require": {
  "aws/aws-sdk-php-laravel": "1.*"
}

設定例は下記です。ここでは、アクセスキーやシークレット、エンドポイントURL を環境変数で渡しています。AWS ではおなじみですが、regionにキューを作成したリージョンを指定するのを忘れないようにしましょう。

    'default' => 'sqs',
    'connections' => [
        'sqs' => [
            'driver' => 'sqs',
            'key' => getenv('AWS_ACCESS_KEY_ID'),
            'secret' => getenv('AWS_SECRET_ACCESS_KEY'),
            'queue' => getenv('AWS_SQS_URL'),
            'region' => 'ap-northeast-1',
        ],
    ],

http://aws.amazon.com/jp/sqs/

ユニットテスト

キューにジョブを登録する側のテストについてです。

QueueクラスをshouldReceiveメソッドでモック化して、テストするのが手軽です。

<?php
use CarbonCarbon;

/**
 * Class QueueTest
 */
class QueueTest extends TestCase
{
    /**
     * @test
     */
    public function queuePush()
    {
        $now = Carbon::create(2014, 8, 13, 12, 34, 56);
        Carbon::setTestNow($now);

        Queue::shouldReceive('connected')->once();
        Queue::shouldReceive('push')->once()->with('MyWorker', ['message' => 'Hello!', 'date' => $now]);
        Queue::shouldReceive('later')->once()->with(10, 'MyWorker', ['message' => 'Delayed', 'date' => $now]);

        $this->client->request('GET', '/queue/push');

        $this->assertTrue($this->client->getResponse()->isOk());
    }
}

サンプルアプリケーション(Heroku)

このエントリの内容を実装したサンプルアプリケーションを github に公開しています。

コード自体はシンプルで、http://localhost/quque/push にアクセスするとジョブがキューに登録されます。あとは php artisan queue:listen コマンドでジョブが実行されます。

Heroku で試せるように、heroku_create というシェルスクリプトで、heroku 関連のコマンドを記述しています。このコマンドを流せば、Heroku アプリケーションが構築されます。

Heroku でのポイントは、`Procfile`で、ワーカープロセスとしてリスナーを起動するという点です。

web: vendor/bin/heroku-php-apache2 public/
worker: php artisan queue:listen

このアプリケーションを Heroku にデプロイすると、web(ジョブをキューに登録する側)、worker(リスナー)の Dyno が構築されます。

あとは Heroku の管理画面で worker の Dyno 数を 1 にすると、web 経由で登録したジョブが worker によって実行されるようになります。

https://github.com/shin1x1/laravel-queue-sample

さいごに

Laravel の Queue について見てきました。

最後に、どのドライバを使うかについてですが、開発環境では beanstalkd を利用するのが手軽で良いでしょう。ローカルにインストールするので動作も速いです。

本番環境では、要件によりけりですが、国内にアプリケーションサーバがあるなら SQS、Heroku など US リージョンを利用するなら IronMQ が良さそうです。

もちろん、自前で beantalkd や Redis を立てるのも良いですが、利用できるなら、ありものを利用するのが楽ですね。

このように、開発環境と本番環境とでドライバを変えても、コード側は一切変更する必要が無いというのは、よく出来ていますね。

参考

blog.ISHINAO.net | Laravel 4でキューを使ってみる
laravel – キュードライバにbeanstalkdを使用する – Qiita

Pocket

follow us in feedly

Home > AWS | Laravel | PHP > Laravel の Queue で非同期処理を実装する(beanstalkd / IronMQ / SQS)

検索
フィード
メタ情報

Return to page top