Resque で学ぶジョブキューイング

client.pl

#!/usr/bin/env perl
use strict;
use warnings;

use Resque;
use Log::Minimal;

my $resque = Resque->new(redis => '127.0.0.1:6379');

print 'input text: ';
while (my $text = <STDIN>){
    chomp $text;
    if ($text) {    # if defined
        $resque->push('echo' => {
                class => 'My::Echo',
                args => [ $text ],
            });
        infof('enqueue: %s', $text);
    }
    print 'input text: ';
}

worker.pl

#!/usr/bin/env perl
use strict;
use warnings;

use Resque;
use Log::Minimal;

my $worker = Resque->new(redis => '127.0.0.1:6379')->worker;
$worker->add_queue('echo');
infof('### start worker');
$worker->work;
infof('### stop worker');

また、ワーカーから実行する処理をモジュールとして作成しておきます。実行時には perform メソッドが呼び出されるので、ここに処理を記述します。

lib/MyTask/Echo.pm

package MyTask::Echo;
use strict;
use v5.10;

use Log::Minimal;

sub perform {
    my $job = shift;
    infof('dequeue: %s', ddf($job->args));
    say $job->args->[0];
}

1;

今回のサンプルで利用している依存モジュールもインストールしておきます。

• Resque
• Log::Minimal

さて、実行してみましょう。

ターミナルを複数用意し、一方で worker.pl を実行します。

$ ./worker.pl
2012-12-12T12:12:02 [INFO] ### start worker at ./worker.pl line 10

そして、もう一方では client.pl を実行し、何かテキストを入力します。

$ ./client.pl
input text: hello
2012-12-12T12:12:07 [INFO] enqueue: hello at ./client.pl line 18

すると、入力したテキストが worker.pl を実行していたターミナルに表示されます。

$ ./worker.pl
2012-12-12T12:12:02 [INFO] ### start worker at ./worker.pl line 10
2012-12-12T12:12:12 [INFO] dequeue: ['hello'] at /Users/hatak/work/jq/lib/MyTask/Echo.pm line 9
hello

worker.pl を複数立ち上げた場合、クライアントから送信したテキストはいずれかのワーカープロセスで処理され表示されます。もちろん、複数のクライアントでも動作します。簡単ですね。

共通

resque:queues

[set型] キューのリスト

resque:workers

[set型] ワーカープロセスのリスト

resque:failed

[list型] 失敗したジョブの結果 (JSON) のリスト

resque:stat:processed

[string型] 取り出した (dequeue) ジョブの総数

resque:stat:failed

[string型] 処理に失敗したジョブの総数

キュー

resque:queue:<キュー名>

[list型] ジョブ (JSON) のリストが格納されているキュー本体、

ワーカープロセス

resque:worker:<ホスト名>:<ポート番号>:<キュー名>:started

[string型] ワーカープロセスの開始時刻

resque:stat:processed:<ホスト名>:<ポート番号>:<キュー名>

[string型] ワーカーが取り出した (dequeue) ジョブの数

resque:stat:failed:<ホスト名>:<ポート番号>:<キュー名>

[string型] ワーカーが処理に失敗したジョブの数

ワーカーは "resque:queue:<キュー名>" からジョブを取り出し、処理します。processed のカウントは取り出した時点で処理の成否にかかわらずインクリメントされます。
処理が失敗した場合は "resque:failed" のリストに JSON シリアライズされて追加され、failed のカウントがインクリメントされます。

これらのデータは、キューやワーカーの状態によってキーの存在自体が変化します。
例えば、ワーカーごとのキーはワーカープロセスを終了すると削除されるので、ワーカーを止めたりして動きを見てみると面白いです。