rabbitmq模式 RPC

rabbitmq模式 RPC

rpc_server.php

<?php

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

$connection = new AMQPStreamConnection('localhost', 5672, 'admin', 'admin','vhost1');
$channel = $connection->channel();

$channel->queue_declare('rpc_queue', false, false, false, false);

function fib($n)
{
if ($n == 0) {
return 0;
}
if ($n == 1) {
return 1;
}
return fib($n-1) + fib($n-2);
}

echo " [x] Awaiting RPC requests\n";
$callback = function ($req) {
$n = intval($req->body);
echo ' [.] fib(', $n, ")\n";

$msg = new AMQPMessage(
(string) fib($n),
array('correlation_id' => $req->get('correlation_id'))
);

$req->delivery_info['channel']->basic_publish(
$msg,
'',
$req->get('reply_to')
);
$req->ack();
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume('rpc_queue', '', false, false, false, false, $callback);

while ($channel->is_open()) {
$channel->wait();
}

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

rpc_client.php

<?php

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

class FibonacciRpcClient
{
    private $connection;
    private $channel;
    private $callback_queue;
    private $response;
    private $corr_id;

    public function __construct()
    {
        $this->connection = new AMQPStreamConnection('localhost',5672,'admin','admin','vhost1');
$this->channel = $this->connection->channel();
list($this->callback_queue, ,) = $this->channel->queue_declare("",false,false,true,false);
$this->channel->basic_consume(
$this->callback_queue,
'',
false,
true,
false,
false,
array(
$this,
'onResponse'
)
);
}

public function onResponse($rep)
{
if ($rep->get('correlation_id') == $this->corr_id) {
$this->response = $rep->body;
}
        }

public function call($n)
{
$this->response = null;
$this->corr_id = uniqid();

$msg = new AMQPMessage(
(string) $n,
array(
'correlation_id' => $this->corr_id,
'reply_to' => $this->callback_queue
)
);
$this->channel->basic_publish($msg, '', 'rpc_queue');
while (!$this->response) {
$this->channel->wait();
}
return intval($this->response);
}
}

$fibonacci_rpc = new FibonacciRpcClient();
$response = $fibonacci_rpc->call(30);
echo ' [.] Got ', $response, "\n";

运行调试

php rpc_server.php

php rpc_client.php