AnyEvent::RabbitMQ - メッセージキュー…使いこなしてますか？

前置き

こんにちは。cooldaemon と申します。みなさん、意識は高まっていますか？ 私は上々です。今回は拙作の AnyEvent::RabbitMQ というモジュールを紹介いたします。

AnyEvent::RabbitMQ は、AMQP ブローカーと非同期にメッセージを送受信するための AMQP クライアントです。AMQP とは、メッセージを扱うミドルウェアのオープンな標準仕様です。AMQP の正確で詳しい内容は、AMQP の公式サイトや Google をご確認ください。

本当の事を言うと、名前は AnyEvent::AMQP::Client でも良かったのですが、私は、AMQP ブローカー実装の一つ RabbitMQ を常用しており、RabbitMQ 以外の AMQP ブローカーを利用する機会も必要もなく、また、接続性を保証する事もしたくなかったので、あえて名前に AMQP を含めずに RabbitMQ を含めました。

で、結局、何ができるの？

率直に申し上げると、RabbitMQ + AnyEvent::RabbitMQ は、Gearman、TheSchwartz、Qudo、Q4M 等のライバルです。

元々、個人的趣味で細々と作成していたのですが、仕事で下記の条件を満たした Web API を作成する必要があり、試しに利用してみたところ、かなり満足の行く結果を得られましたので、今回は、その事例を中心に紹介いたします。

• 秒間 200 件程度の HTTP リクエストを捌く必要がある
• 外部システムとの接続の都合で、一つの HTTP リクエストは、平均 5 秒は接続し続ける必要がある
• 既存の API と同一のドメインを利用する必要がある

同時接続 1000 程度の HTTP リクエストなら、社内に転がっている 1U サーバ*1一台で十分捌けるだろうし、機器故障に備えて二重化しても二台で十分だろうという予測の元、下記の構成でシステム設計を行いました。

[様々な HTTP クライアント] <-> [Apache * N台] <-> [RabbitMQ * 2台] <-> [Perl 製の Worker * 2台]

Apache と Worker の間の RabbitMQ の役割は、ここの 6.RPC の図の通りです。

やっと本題

Perl で記述した HTTPD Worker の例は、下記の通りです。 まずは、ソースコード全体を俯瞰してください。

#!/usr/bin/env perl

use strict;
use warnings;

use Coro;
use Net::RabbitFoot;
use JSON::XS;
use Readonly;
use Data::Dumper;

Readonly my @MQ_CONNECT_ARGS => (
    host  => '10.0.0.10',
    port  => 5672,
    user  => 'guest',
    pass  => 'guest',
    vhost => '/',
);

Readonly my $PREFETCH => 5;

$| = 1;

$SIG{HUP} = $SIG{INT} = $SIG{TERM} = sub {
    warn "Trapped SIGNAL.\n";
    $Coro::main->ready;
};

my $json = JSON::XS->new()->utf8();

my $rf = eval {
    Net::RabbitFoot->new()->load_xml_spec()->connect(
        @MQ_CONNECT_ARGS,
        (map {
            'on_' . $_ => failure_handler('connecn.on_' . $_)
        } qw(close read_failure)),
    );
};
die $@ if $@;

eval {
    my $ch = $rf->open_channel(
        on_close => failure_handler('channel.on_close'),
    );
    $ch->qos(prefetch_count => $PREFETCH);
    $ch->consume(
        queue      => 'http',
        no_ack     => 0,
        on_consume => unblock_sub {
            work($json, $ch, shift);
        }
    );
};
if ($@) {
    $rf->close;
    die $@;
};

schedule;
$rf->close;
exit;

sub failure_handler {
    my ($event) = @_;
    return unblock_sub {
        warn Dumper({$event => \@_});
        $Coro::main->ready;
    };
}

sub work {
    my ($json, $ch, $request,) = @_;

    my $response = make_http_response(@_);

    $ch->publish(
        routing_key => $request->{header}->reply_to,
        header      => {
            app_id  => $response->{code},
            headers => $response->{headers},
        },
        body        => $response->{body},
        on_return   => failure_handler('channel.on_return'),
    );

    $ch->ack(
        delivery_tag => $request->{deliver}->method_frame->delivery_tag,
    );

    return;
}

sub make_http_response {
    my ($json, $ch, $request,) = @_;

    warn Dumper({headers => $request->{header}->headers});

    my $request_body = $json->decode($request->{body}->payload);

    warn Dumper({body => $request_body});

    return {
        code    => 200,
        headers => {
            'Content-Type' => 'text/plain; charset=UTF-8',
        },
        body    => q{Merry X'mas},
    };
}

*** RabbitMQ へ接続して Channel を開く

スレッド毎に AMQP ブローカーと接続する行為は、リソースの無駄遣いであるため、AMQP には、一つの接続を Channel で分割する仕様が定義されています。通常は、各スレッド毎に Channel を開く事でスレッドセーフを保証します。

今回は、継続による疑似スレッドという事もあり、同時に実行されるスレッドは常に一つである事から、キューの監視で使用している Channel を、各スレッド内のメッセージ送信時にも使い回しています*5。もし Coro ではなく ithread を使う際には、スレッド毎に Channel を開く必要があります。

*** キューに貯まったメッセージを 5 個ずつ捌く事を RabbitMQ へ通知

qos で設定した prefetch_count がそれです。ここに 5 と指定すると、この Channel の処理中のメッセージが 5 個になるまで、メッセージが届く度に on_consume の unblock_sub が呼ばれます。つまり、ここに指定した値が並列度となります。

*** キューの監視を行うサブルーチンを登録

Coro から利用する場合は、必ず unblock_sub を指定してください。イベントループからコールバックサブルーチンが呼ばれるので、このコールバックサブルーチン内で Coro::rouse_cb を利用する際に具合が悪いのです。

no_ack に 0 を指定しているので、Ack しないとメッセージは消費されませんが、ここは 1 にして Ack 不要にしても構いません。

*** シグナルや障害が発生するまで待つ

schedule しているだけです。つまり、初期化処理後の main スレッドの主な仕事は、大人しく静かに寝ている事です。$Coro::main->ready が呼ばれるという事は、この Worker の終了を意味します。

*** HTTP Request Header を標準エラーに出力

on_consume に指定されたコールバックサブルーチンの第一引数はハッシュとなっており、header と body というキーを持ちます。メッセージを送る publish の引数と比較して頂けると解り易いのですが、$request の header と body には、拙作の rabbitmq-http-proxy が Publish 時に指定した header と body が入っています。

*** HTTP Request Body を標準エラーに出力

rabbitmq-http-proxy の癖で、GET/POST されたパラメータは JSON 化されて body にセットされています。そこで、JSON::XS で decode しています。

*** HTTP Response を返信用のキューに送信

これまた rabbitmq-http-proxy の癖で、header の app_id に HTTP Status Code を指定しています。headers が配列のリファレンスではなく、ハッシュのリファレンスな所も、どうにかしたい所です。実案件では、headers に CGI.pm で作成した Cookie 等の HTTP Headers を設定しています。

*** メッセージを Ack する

前述しましたが、Consume 時の no_ack に 0 を指定しているので、メッセージを消費する為に Ack が必要です。Ack しない場合、Consume している Channel が閉じるか Cancel するまで、この Worker の別スレッドや、他の Worker が同一のメッセージを受け取る事ができません。

以上で Worker の説明終わりです。簡単でしたか？難しかったでしたか？簡単だと思って頂ければ成功なのですが…。