RabbitMQ 入门指南（六）

小灰灰 2022-12-27 01:14 175阅读 0赞

## Remote Procedure Call (RPC) ##

Remote Procedure Call，即远程过程调用，简称 **RPC**，它是一种网络数据交互的协议规范，它让客户端能够像调用本地的函数一样调用远程服务端的函数。

多系统之间的内部数据交互，一般采用 RPC。

在 RabbitMQ 入门指南（二）中，我们学习了如何使用 Work Queue 将耗时的任务分配给多个 worker。

但是，如果我们需要调用远程服务器的方法并等待它的返回结果，该怎么处理？

这里，我们将使用 RabbitMQ 来构建一个十分简单的 RPC 系统（它包含一个客户端和一个可扩展的 RPC 服务端）。

### Client interface ###

为了展示 RPC 服务如何使用，我们将创建一个简单的 client class。在该类中，有一个 call 方法，用于发送 RPC 请求，该方法会阻塞直到收到服务端的响应。

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

**说明：** 尽管 RPC 是非常常见的网络通信模式，但它却是饱受争议的。

### 回调队列 ###

通常，在 RabbitMQ 上执行 RPC 是很容易的。客户端发送请求消息，服务器返回响应消息。为了接收响应，客户端在发送请求时需要指定 callback 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 协议预先定义了消息的 14 种属性，大多数属性很少用到，除了下面四种：

*  delivery\_mode：标记消息是否是持久化的。2 表示持久化的，1 表示短暂的。
 *  content\_type：用于描述 mime-type 数据编码类型，如 json 类型的数据使用 application/json。
 *  reply\_to：通常用于指定 callback queue。
 *  correlation\_id：用于设置每个 RPC 请求的唯一标识，从而关联 RPC 响应和请求。

### correlation\_id ###

在上面的代码中，我们会为每一个 RPC 请求创建一个回调队列，这很低效。幸运的是，我们只需为每一个客户端创建一个回调队列。

这将导致新的问题：在该回调队列中，我们不清楚收到的响应属于哪一个请求？这时，correlation\_id 就派上用场了，每一个请求都会有一个唯一 correlation\_id 值。

## 总结 ##

![image][]

我们的 RPC 系统工作流程大致如下：

*  启动客户端，创建一个匿名的、排他的回调队列。
 *  对于一次 RPC 请求，客户端发送一条消息，该消息有两个属性 reply\_to 和 correlation\_id。
 *  把请求投递到 rpc\_queue 队列中。
 *  RPC worker 进程等待接收处理队列中的请求消息。当它接收到请求消息时，处理该任务，然后将结果消息投递到回调队列中。
 *  客户端等待接收回调队列中的响应结果消息。当接收到消息时，会先检查 correlation\_id 值是否匹配，只有当该消息的 correlation\_id 和发送请求时的 correlation\_id 完全匹配时，才将最终的结果返回给应用。

## 最终的代码示例 ##

rpc\_server.php，作为 RPC 服务端。

<?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 ($channel->is_consuming()) {
        $channel->wait();
    }
    
    $channel->close();
    $connection->close();

rpc\_client.php，作为 RPC 的客户端。

<?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,
                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";

现在，一个简易的 RPC 服务已经搭建好了。

# 启动 RPC 服务端
    php rpc_server.php
    # => [x] Awaiting RPC requests

# 启动 RPC 客户端
    php rpc_client.php
    # => [x] Requesting fib(30)

**说明：**

上面我们只是展示了实现 RPC 服务的一种可能的方式，非常简易，无法直接用于生产环境，只是抛砖引玉。但它却有两个非常重要的优点：

*  可扩展性强。如果 RPC 服务端响应太慢，只需多开几个 RPC 服务进程。
 *  对于 RPC 客户端，只需发送和接收一条消息。因此，对于单个 RPC 请求，只需一次网络往返通信。

**参考文献**

\[1\] [https://www.rabbitmq.com/tutorials/tutorial-six-php.html][https_www.rabbitmq.com_tutorials_tutorial-six-php.html]

[image]: /images/20221120/f9b54568b0c041ba9708ca79ad48452e.png
[https_www.rabbitmq.com_tutorials_tutorial-six-php.html]: https://www.rabbitmq.com/tutorials/tutorial-six-php.html