利用RabbitMQ来做一个RPC的客户端和服务端。

输出斐波那契数列为例子。

rpc_server.php:

<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('192.168.33.10', 5672, 'admin', 'admin');
$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->delivery_info['channel']->basic_ack(
        $req->delivery_info['delivery_tag']
    );
};

//为了在多个服务上面平均负载
$channel->basic_qos(null, 1, null);
$channel->basic_consume('rpc_queue', '', false, false, false, false, $callback);

while (count($channel->callbacks)) {
    $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(
            '192.168.33.10',
            5672,
            'admin',
            'admin'
        );
        $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,
            false,
            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";

运行服务器端，在运行客户端，可以收到计算结果。

几个关键点：

运行服务端，服务端声明了一个RPC队列rpc_queue，然后监听上面的消息。

运行客户端，客户端声明了一个临时队列，并在临时队列上面添加了回调函数，当有响应的时候取出结果。

客户端往RPC队列里面添加RPC调用请求，服务端收到后，把计算结果推送到客户端建立的临时队列里面去，客户端输出结果。

客户端发送的消息里面有两个关键属性reply_to和correlation_id。

replay_to属性告诉服务端程序把结果写到哪个队列去，correlation_id是一个消息的唯一标识，用来区分RPC请求和响应，所以在上面例子onResponse回调里面有个比较消息id的操作，用来确认是客户端发送的消息。

如果每个请求都建立一个临时队列好像有点浪费，那么是不是可以创建一个专门的回调队列呢？因为每个消息是有id的，客户端是可以区分，所以所有请求是可以共用一个回调队列的。

参考 https://www.rabbitmq.com/tutorials/tutorial-six-php.html