AMQP/RabbitMQ めも

loop_reader.php

<?php

/** 兎に角ループさせる */
while (true) {
    echo " [x] If you wanna stop, you push Ctrl-C","\n";
    require("reader.php");
    sleep(1);
}

memo.md

RabbitMQ メモ

RabbitMQとは（ｒｙ）です。

How do i install ?

なんか色々入る。erlang 使ってるし。

Mac なら

  $ brew install rabbitmq
  $ brew install rabbitmq-c

PHP で使う時、MAMP 使ってるなら、 https://github.com/majksner/php-amqp-mamp を使うのが速い。インストール方法も書いてる。

ただコマンドラインでしか使わない時は pecl で。

CentOS なら

  $ yum install rabbitmq-server
  $ yum install librabbitmq-devel

CentOS でPHP使うなら

  $ pecl install amqp

で入ったと思う。入ったら、php.ini に

  extension=amqp.so

を忘れずに。

python 使ってるなら、pipいれてると思うし簡単。

  $ pip install pika==0.9.5

ドキュメントに合わせたバージョン。最新版使いたいーなら、==0.9.5を省けばいい。

どう使うの？

Python は公式ドキュメントおよびサンプルがスマートにスッキリと綺麗に気持ちよく愉快に動きてくれる。

PHPで手間取ったからこのメモある。

sender.php, reader.php, loop_reader.php というのがサンプル。

1. sender.php

RabbitMQ に適当なメッセージを指定した個数だけ追加させる

2. reader.php

RabbitMQからデータを一つだけ取り出して表示して終わる。

3. loop_reader.php

reader.php を1秒間に一回実行させる

実行手順

順番バラバラでも動くけど、とりあえず。

1. php sender.php 120 と実行させ適当なメッセージを貯めさせておく。
2. ターミナルを複数開き、php loop_reader.php と打てば、それぞれが別々のメッセージを取得してるのが観測出来る

いまだによく分からんこと

AMQPChannel って何であるの…

reader.php

<?php
/** php reader.php
 *
 * AMQP 経由でメッセージを一件のみ読み、表示させる
 */

$connection = new AMQPConnection();
$connection->setHost("localhost");
$connection->setPort(5672);
$connection->setLogin("guest");
$connection->setPassword("guest");

if ($connection->connect()) {
    // ok
}
else {
    throw new Exception("Cannot connect AMQP Server.");
}

$channel = new AMQPChannel($connection);
$queue = new AMQPQueue($channel); 
$queue->setName("task_queue");
$queue->setArgument("durable", 1);
$msg = $queue->get();

if ($msg !== false) {
    echo " [x] Message: ", $msg->getBody(),"\n";
    $queue->ack($msg->getDeliveryTag());
}

$connection->disconnect();

sender.php

<?php
/** php sender.php
 *
 * 引数で数を指定した分だけ、ランダム文字を作成してAMQP経由でRabbitMQにデータを入れる
 */

$cnt = 0;
if (isset($argv[1]) && is_numeric($argv[1]))
    $cnt = $argv[1];

if ($cnt === 0) {
    echo "php sender.php <cnt>", "\n";
    echo "  引数で数を指定した分だけ、ランダム文字を作成してAMQP経由でRabbitMQにデータを入れる","\n";
    exit(0);
}

function randomString($nLengthRequired = 20, $sCharList="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_"){
    mt_srand();
    $sRes = "";
    for($i = 0; $i < $nLengthRequired; $i++){
        $sRes .= mb_substr($sCharList, mt_rand(0, mb_strlen($sCharList) -1), 1);
    }
    return $sRes;
}

$connection = new AMQPConnection();
$connection->setHost("localhost");
$connection->setPort(5672);
$connection->setLogin("guest");
$connection->setPassword("guest");

if ($connection->connect()) {
    // ok
}
else {
    throw new Exception("Cannot connect AMQP Server.");
}

$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);

for ($i = 0; $i < $cnt; $i++) {
    $data = randomString();
    $exchange->publish($data, "task_queue", AMQP_NOPARAM, array('delivery_mode'=>2));
    echo " [o][",$i,"] add message ","\n";
}


$connection->disconnect();