八、远程调用（RPC）

在云计算环境中，很多时候需要用它其他机器的计算资源，把一部分计算任务分配到其他节点来完成。RabbitMQ 如何使用 RPC 呢？下面将会通过其它节点完成斐波纳契示例。

流程图

• 当客户端启动时，它创建了匿名的 callback queue
• 客户端发起 RPC 请求时将同时设置两个 properties：reply_to 设置为 callback queue；correlation_id 设置为每个请求都是独一无二的值
• 请求将被发送到一个 rpc_queue 队列
• RPC 端或者说 Server 一直在等待那个 queue 的请求，当请求到达时，它将通过在 reply_to 指定的 queue 回复一个 Message 给客户端
• client 一直等待 callback queue 的数据，当 Message 到达时，它将检查 correlation_id 的值，如果值和它 request 发送时的一致那么就将返回响应

演示代码

rpc_client.php

<?php
/**
 * RPC客户端
 */
$routing_key = 'rpc_queue';
$num = empty($argv[1]) ? 0 : intval($argv[1]);

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);

$client_queue = new AMQPQueue($channel);
$client_queue->setFlags(AMQP_EXCLUSIVE);
$client_queue->declareQueue();
$callback_queue_name = $client_queue->getName();

$corr_id = uniqid();
$properties = [
    'correlation_id' => $corr_id,
    'reply_to' => $callback_queue_name
];

$exchange = new AMQPExchange($channel);
$exchange->publish($num, $routing_key, AMQP_NOPARAM, $properties);

$client_queue->consume(function($envelope, $queue) use ($corr_id){
    if ($envelope->getCorrelationId() == $corr_id) {
        $msg = $envelope->getBody();
        var_dump('Received Data: ' . $msg);
        $queue->nack($envelope->getDeliveryTag());
        return false;
    }
});

// 断开连接
$connection->disconnect();

rpc_server.php

<?php
/**
 * RPC服务端
 */
$routing_key = 'rpc_queue';

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);
$channel->setPrefetchCount(1);

$server_queue = new AMQPQueue($channel);
$server_queue->setName($routing_key);
$server_queue->declareQueue();

$exchange = new AMQPExchange($channel);

$server_queue->consume(function($envelope, $queue) use ($exchange){
    $num = intval($envelope->getBody());
    $response = fib($num);
    $exchange->publish($response, $envelope->getReplyTo(), AMQP_NOPARAM, [
        'correlation_id' => $envelope->getCorrelationId(),
    ]);
    $queue->ack($envelope->getDeliveryTag());
});

// 断开连接
$connection->disconnect();

// 斐波那契函数
function fib($num) {
    if ($num == 0)
        return 1;
    else if ($num == 1)
        return 1;
    else
        return fib($num - 1) + fib($num - 2);
}

测试示例：