0

だから、私は 4 つのことを行う単純なアプリを作成しようとしています。

1）消費者のリストを取得します（登録されたものを取得できれば..参加するときに何らかの名前を付けることができるので、動的です）。

2) 1 つのランダムなコンシューマに「メッセージ」を送信し、結果を表示する

3) 1 人の特定の消費者 (上記のリストまたは事前定義されたリストから取得) に「メッセージ」を送信し、結果を表示します。

4) すべてのコンシューマに「メッセージ」を送信し、それぞれからの結果を表示します。

アプリは php-amqplib ( https://github.com/videlvaro/php-amqplib ) を使用して php で実行されます。rabbitmq は稼働中で、動作しているようです (チュートリアルを試してみました)。

私にとってamqp libのドキュメントは少し奇妙ですので、サンプルコードのいくつかの行と使用されているパラメータの説明をいただければ幸いです。

4

1 に答える 1

1

1) は、rabbitmq の管理 /api/queues を使用して、各コンシューマーに永続的なキュー名を付けることで解決できます。

2,3,4 は、次のように解決されました。

<?php
//Producer Config
$host = "remote_host";
$port = 5672;
$user = "user";
$pass = "pass";
$array = array("consumer1", "consumer2");

...

<?php
//Producer

if(!isset($argv[1]) || !isset($argv[2])){
        die("Specify a target and a message\n");
}

require_once __DIR__.'/config.php';
require_once __DIR__.'/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPSSLConnection;
use PhpAmqpLib\Message\AMQPMessage;

$corr_id = uniqid();

$connection = new AMQPSSLConnection($host, $port, $user, $pass, "/", array("verify_peer" => false));
$channel = $connection->channel();

$response = null;
$onResponse = function ($rep) {
        global $response;
        $response = $rep->body;
        echo " [>] Received: '".$response."'\n";
};

list($callback_queue, ,) = $channel->queue_declare("", false, false, true, false);
$channel->basic_consume($callback_queue, '', false, false, false, false, $onResponse);
$msg = new AMQPMessage($argv[2], array('correlation_id' => uniqid(), 'reply_to' => $callback_queue));

switch($argv[1]){
        case "random":
                $dest = $array[array_rand($array)];
                $type = "direct";
                break;
        case "all":
                $dest = "to_all";
                $type = "fanout";
                break;
        case $argv[1]:
                $dest = $argv[1];
                $type = "direct";
                break;
}

$channel->exchange_declare($dest, $type, false, false, false);
$channel->basic_publish($msg, $dest);
echo " [<] Sent '".$argv[2]."' to '".$dest."'\n";

try {
        if($dest == "to_all"){
                $replies = 0;
                while(!$response || $replies < count($consumers_array)){
                        $channel->wait(null, false, $timeout);
                        $replies++;
                }
        }else{
                while(!$response){
                        $channel->wait(null, false, $timeout);
                }
        }
}catch(PhpAmqpLib\Exception\AMQPTimeoutException $e){
        echo " [x] AMQPTimeoutException thrown\n";
}

$channel->close();
$connection->close();

...

<?php
//Consumer config
$host = "remote_host";
$port = 5672;
$user = "user";
$pass = "pass";
$consumer_name = "consumerX";

...

<?php
//Consumer
require_once __DIR__ . '/config.php';
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPSSLConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPSSLConnection($host, $port, $user, $pass, "/", array('verify_peer' => false));
$channel = $connection->channel();

$channel->exchange_declare('to_all', 'fanout', false, false, false);
$channel->exchange_declare($consumer_name, 'direct', false, false, false);
$channel->queue_declare($consumer_name, false, false, true, false);
$channel->queue_bind($consumer_name, 'to_all');
$channel->queue_bind($consumer_name, $consumer_name);
echo '[*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg) {
        echo " [>] Received: '".$msg->body."'\n";
        $rand = rand(1,3);
        sleep($rand);
        $reply = uniqid()." - slept ".$rand;
        echo " [<] Replied: '".$reply."'\n";
        $raspuns = new AMQPMessage($reply);
        $msg->delivery_info['channel']->basic_publish($raspuns,'',$msg->get('reply_to'));
};

$channel->basic_consume($consumer_name, '', false, true, false, false, $callback);

while(count($channel->callbacks)) {
        $channel->wait();
}

$channel->close();
$connection->close();

これが最善の方法かどうかはわかりませんが、「Hello World」がうまくいきました。

于 2016-01-23T22:20:36.373 に答える