技術標籤：訊息佇列中介軟體rabbitmq

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 值。

總結

我們的 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