本文所有内容均个人从RabbitMQ官网教程中翻译，若图片文字的引用有任何侵权的地方，联系我，我会立马删除。
This article was translated from RabbitMQ Official Tutorials by myself，and if this article and the images in this article have any infringement,please contact to me, and i will delete them.
远程程序调用（Remote Procedure Call(RPC)）
（使用php-amqplib）
在第二个教程中我们学习了使用工作队列（Work Queues）在多个工作程序中分配耗时任务。
但是如果我们需要在远程计算机上运行一个函数并等待它的结果呢？那是另外一回事了。这种模式通常被成为远程程序调用（Remote Procedure Call）或者简称为PRC。
在这一教程我们准备使用RabbitMQ来建立一个RPC系统：一个客户端和一个可扩展的PRC服务器。由于我们没有一些值得分发的耗时任务，我们准备建立一个会返回斐波那契数字（Fibonacci）的愚蠢的（dummy？）RPC服务。
客户端接口
为了说明一个RPC服务能被如何使用，我们准备创建一个简单的客户端类。它会暴露一个叫做call的方法用于发送一个RPC请求，然后一直阻塞直到接收到应答：
$fibonacci_rpc = new FibonacciRpcClient();
$response = $fibonacci_rpc->call(30);
echo " [.] Got ", $response, "\n";
关于RPC的一点
尽管RPC是计算中一个十分普通的常见的模式，但它常常被批评。当程序员没有意识到一个函数是本地调用还是一个缓慢的RPC（请求）时候问题就会出现。这样的困惑导致了一个不可预知的系统以及为调试增加比必要的复杂性。和简化软件不同，滥用RPC将会导致代码像一团意大利面一样难以维护。
将下面的建议铭记于心：
- 确保能显而易见地区分哪个函数是本地调用哪个是远程。
- 为你的系统书写文档。确保组件间的依赖关系清晰。
- 处理错误情况。当RPC服务器挂掉了很长的一段时间，客户端应该如何做出反应？
有疑问时应避免RPC。如果可以，你应该使用一个异步的管道——而不是像会造成阻塞的这样的RPC，结果会被异步地推送到下一个计算阶段。
回调队列（Callback Queue）
一般情况下，实现通过RabbitMQ实现RPC是很容易的。一个客户端发送一个请求消息然后一个服务端回复一个响应消息。为了去接收一个响应，我们需要随请求发送一个“回调”队列地址。我们使用默认Queue（队列）。让我们一起来试试吧：
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
$msg = new AMQPMessage(
$payload,
array('reply_to' => $queue_name));
$channel->basic_publish($msg, '', 'rpc_queue');
# ... then code to read a response message from the callback_queue ...
消息属性
AMQP 0-9-1协议的消息预定义了一套十四个属性。这些属性的大部分都很少使用，除了以下的这些：
delivery_mode：标记一条消息为持久化的（设为2时）或者瞬时性的（设为1时）。你可能还记得我们在第二个教程中由使用过。content_type：用来描述编码的mime-type。例如对于经常使用的JSON编码，它是一个用来将这个属性设为application/json的很好的一个实例。reply_to：通常被用来命名一个回调队列。correlation_id：对于RPC的响应与请求间的关联很有用。
关联ID（Correlation ID）
上面提及的方法中我们建议为每一个RPC请求都创建一个回调队列。这是十分低效率的，但幸运的是，这里有一个更好的方法，让我们为每个客户端创建一个单一的回调队列。
这引发了一个新问题，在一个队列中接收到一个响应后，并不清楚这个响应应该属于哪一个请求。这正式使用correlation_id属性的时候。我们准备把这个值对于每一个请求来说都是唯一的。稍后，当我们从回调队列接收到一个消息我们会查看这一属性，然后根据它我们将能够匹配请求到对应的响应。如果我们看见了一个未知的correlation_id值，我们可以安全地丢弃这一消息，因为它不属于我们的任何一个请求。
你可能会问，在回调队列中，为什么我们要忽视未知的消息，而不是抛出异常？因为这可能是服务器的竞争情况。尽管不太可能，这有可能是RPC服务器在发送回应答给我们后就立马挂掉了，但这却是在为请求发送一个确认消息前。如果这一情况发生，重启后的RPC服务器将会再次处理请求。这就是为什么在客户端中我们必须优雅地处理重复响应，而RPC在理想情况中应该是幂等的。
总结
RPC system
我们的RPC将会像这样子工作：
当客户端Client开始运行，让创建了一个匿名的独占回调队列。
对于一个RPC请求，客户端Client发送一条带有两个属性的消息：设置回调队列的reply_to与对于每个请求都是唯一的值的correlation_id。
请求被发送到一个rpc_queueQueue（队列）。
RPC处理程序（又名服务）将会等待来自该队列的请求。当一个请求出现，它会处理工作并通过reply_to参数指明的Queue（队列）发送一条带有结果的消息回服务端Client。
客户端将会在回调队列等待数据。当一条消息出现，它会检查correlation_id属性。如果它匹配了请求的值，它将会返回响应到应用。
将他们一起运行
斐波那契（Fibonacci）任务代码：
function fib($n) {
if ($n == 0)
return 0;
if ($n == 1)
return 1;
return fib($n-1) + fib($n-2);
}
我们声明我们的斐波那契函数。它只接收有效的正整数输入。（不要期望这一个函数处理很大的数，这可能是最慢的递归实现了。）
我们的RPC服务rpc_server.php的代码如下：
<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$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();
?>
服务端代码同样十分容易理解：
像往常一样我们建立了连接、频道与定义了队列。
我们可能希望运行多于一个的服务进程。为了将负载均衡地分布在多个服务，我们需要在$channel.basic_qos方法中设置prefetch_count设置参数。
我们使用bask_consume函数去访问队列。然后我们进入了呆呆请求消息的while循环，处理重做然后发会响应。
我们的RPC客户端代码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, 'guest', 'guest');
$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, 'on_response'));
}
public function on_response($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_client.php与rpc_server.php。
我们的RPC服务现在已经准备好了。我们可以开启服务：
php rpc_server.php
# => [x] A waiting PRC requests
为了请求以个斐波那契数字，我们运行客户端：
php rpc_client.php
# => [x] Requesting fib(30)
这里介绍的设计并不是RPC服务的唯一实现，但它有一些重要的优点：
如果RPC服务非常缓慢，你可以通过运行另外一个服务来扩展。尝试一下运行第二个rpc_server.php在一个新的控制台。
在客户端，RPC只需要发送和接收单一消息。不需要类似queue_declare的异步调用。因此RPC的每一个RPC请求都只需要一次网络往返。
我们的代码仍然十分简单，且没有尝试去解决更复杂（却很重要）的问题，例如：
- 如果没有服务正在运行，客户端应该如何反应。
- RPC客户端应该由超时机制吗？
- 如果服务器发生故障或者异常，是否将其转发给客户端？
- 在处理前防止无效的传入消息（如检查边界、类型）。
如果你想进行实验，可能会发现一个UI管理对于查看队列非常有用
网友评论