TheSchwartzで仕事をあとにまわす

web2.0 時代のジョブキューサーバー Gearman と TheSchwartz の関係について - TokuLog 改めB日記に書かれているとおり、Gearmanは仕事を投げられたらすぐやって返す前提になっていて今やりたくないけどあとでやるみたいなのができません。

たとえば、10分後にならできるんだけど、という仕事が来たとします。
このときGearmanのワーカの中でsleepして10分待つと、後から来たほかの仕事を一切しないで10分待つことになってしまって、後から来た仕事が今すぐできるものだったとしても10分待たれさるのでGearmanだと今来た仕事のためにsleepして待つわけにはいきません。

結果としてGearmanだとやってきた仕事を今すぐやるか、絶対やらないか、の二択になってしまいます。

それだと困るので、もう一方のTheSchwartzだと、今やりたくないけどあとでやる、ことができるのかなーと思って調べてみたらできました。GearmanとくらべてTheSchwartzのほうはどうやって使うかあまり書かれてなかったので、TheSchwartzを使ってやってきた仕事を後に回す方法を書きます。

MySQLにテーブルを作る

Gearmanが仕事のキューをメモリ上に持つのに対して、TheSchwartzはデータベースに保存します。だからはじめにデータベースにテーブル作ったりするんだろうなと思ったのですがTheSchwartz - reliable job queue - search.cpan.orgにもTheSchwartz をためす - TokuLog 改めB日記にもテーブルを作ることについて触れられていません。

なければ勝手に作ってくれるのかなと思ってSYNOPSISのコードを実行してみたけどやっぱり自動でできたりはしなそうだったので調べたらCatalyst and TheSchwartz: Reliable JobQueue in a great framework - Voxにschema.sqlを使うといいと書いてありました。このスキーマをmysqlで実行すればTheSchwartzのキューを管理するためのテーブルが出来上がります。

コード

セットアップができれば後はGearmanとおなじかんじです。workerで仕事の名前を登録して、仕事を待ちます。 Gearmanとちがうところは、関数を登録するのではなくTheSchwartz::Workerのサブクラスを作って、そのクラス名を登録するところ。仕事が来た時には作ったサブクラスのworkメソッドが呼び出されます。

TheSchwartzのサーバは、デフォルトでは5秒間隔でキューをチェックして、仕事があった時にはその仕事の名前に応じたワーカを起動します。workはふられた仕事に対してWORKINGにある4つのメソッドのうちのどれかを呼び出して終了します。

completed()

仕事を問題なくこなした時に呼び出します。TheSchwartzでは問題なく終わった時は、その結果をどこかに保存したりはできないようで、なにも引数はありません。そういうときはわざとfailしたことにするのかな。

failed($msg, $exit_status)

やったけどだめだったときに呼び出します。failedを呼ぶと、失敗した時刻、ジョブのID, $msg, $exit_statusがdbのerrorテーブルとexitstatusテーブルに記録されます。perlの実行時エラーでなどで失敗した時も同様にerrorテーブルに記録されるようになっています。

max_retriesが0以上に設定されている時はfailedを呼んでもジョブはキューから削除されずに、max_retriesの値に達するまで即時再度実行されるようになっています。常に失敗するような状態になっている時にmax_retriesが大きな値になっているとすごい勢いでループしてCPUを浪費するので注意が必要です。

permanent_failure( $msg, $exit_status )

リトライしてもだめなエラーの時にはpermanent_failureを呼び出すとmax_retriesに関わらずリトライされません。$exit_status = 1で呼び出すとさらにキューからジョブを削除されます。

replace_with

今のジョブをキューから削除してほかのジョブで置き換えたい時に呼び出します。たとえば、ログインしないとアクセスできないページのデータを読み込みたい時、持っているクッキーがまだ有効だと思ってアクセスしたらクッキーが無効でログインが必要になった場合、ページを読み込むといういまのジョブを

ログインページにアクセス
アカウント情報をpost
データを読み込みたいページにアクセス

という3つのジョブと入れ替えたい、というような時に使います。

今回はこの最後のreplace_withを使って今きた仕事を明日(24時間後)にするサンプルを作りました。
TheSchwartzには時間を指定してジョブを実行させることができるようになっていたりして、やってきた仕事を今すぐやるかやらないかしかないGearmanにくらべて柔軟な管理ができます。

Gearmanに比べるとスペルが覚えられないのとタイプしにくいのが難点です。

worker.pl

package MyWorker;
use base qw( TheSchwartz::Worker );

sub work {
    my ($class, $job) = @_;

    my $deferredJob = TheSchwartz::Job->new(
        funcname => $job->funcname,
        arg => $job->arg,
        run_after => time + 24 * 3600
    );

    $job->replace_with( $deferredJob );
}

package main;
use TheSchwartz;

my $client = TheSchwartz->new(
    databases => [{dsn => 'dbi:mysql:the_schwartz', user => 'root', pass => ''}],
    verbose => 1
);
$client->can_do('MyWorker');
$client->work();

client.pl

use TheSchwartz;
use JSON;

my $client = TheSchwartz->new(
    databases => [{dsn => 'dbi:mysql:the_schwartz', user => 'root', pass => ''}],
    verbose => 1
);

my $arg = {
    hello => "world",
    fizz => "buzz",
};
$client->insert('MyWorker' => to_json($arg) );