【譯】RabbitMQ：遠程過程調用（RPC）

在教程二中，我們學習了如何使用工作隊列在多個工作線程中分發耗時的任務。但如果我們需要去執行遠程機器上的方法並且等待結果會怎么樣呢？那又是另外一回事了。這種模式通常被稱為遠程過程調用（RPC）。

本教程中我們將使用RabbitMQ構建一個遠程過程調用系統：一個客戶端和一個可擴展的服務器。由於沒有什么耗時的任務值得分發，我們將創建一個虛擬的RPC服務用於返回斐波那契數列。

客戶端接口

為了闡釋如何使用RPC服務我們將創建一個簡單的客戶端類。類中獎公開一個方法用於發送一個RPC請求，然后阻塞知道收到應答，方法名稱叫做call:

var rpcClient = new RPCClient();

Console.WriteLine(" [x] Requesting fib(30)");
var response = rpcClient.Call("30");
Console.WriteLine(" [.] Got '{0}'", response);

rpcClient.Close();

 

RPC注記

盡管RPC在計算機技術中是一種非常常見的模式，但是它卻飽受批判，問題發生在程序員不知道一個調用是本地的還是一個耗時的RPC。這樣的混亂，導致不可預知的系統，並將不必要的復雜性調價到調試過程中。誤用RPC將導致不可維護的混亂的代碼，而不是簡化軟件。

銘記這些限制，考慮下面的建議：

• 確保方法是本地調用還是遠程調用能清晰明了
• 將系統歸檔備案，使組件間的依賴關系足夠清晰
• 捕獲異常，當RPC服務宕機很長時間客戶端作何響應？

應該在不能確定的時候避免使用RPC，如果可以的話，你可以使用異步管道，而不是類RPC的阻塞，結果被異步推送到下一個計算階段。

 

回調隊列

一般來說，在RabbitMQ之上構建RPC非常的容易，客戶端發送請求消息，服務返回應答消息。為了能夠接收到應答的消息，我們需要在請求時指定一個回調隊列地址：

var corrId = Guid.NewGuid().ToString();
var props = channel.CreateBasicProperties();
props.ReplyTo = replyQueueName;
props.CorrelationId = corrId;

var messageBytes = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
                     routingKey: "rpc_queue",
                     basicProperties: props,
                     body: messageBytes);

// ... 然后是從回調隊列中讀取消息的代碼 ... 

消息屬性

AMQP協議預定義了一個包含14個屬性的屬性集作用於消息之上，大多數都很少使用，除了下面這些：

• deliveryMode：將消息標記為持續（使用數值2）或瞬時（其他任意值）的，通過教程二你應該還記得這個屬性。
• contentType：用於描述媒體類型編碼，例如：針對常用的JSON編碼，最好的做法是把這個屬性設置為：application/json。
• relayTo：通常用於命名一個回調隊列。
• correlationId：關聯RPC請求和響應的時候非常有用

關聯ID

在上面准備的方法中，我們建議為每一個RPC請求創建一個回調隊列。這樣相當低效，辛運的是有更好的方法，讓我們為每一個客戶端創建一個回調隊列。

這樣引出了一個新問題，當收到一個響應的時候，它無法清楚的知道響應屬於哪一個請求。這就是correlationId派上用場的時候。我們將為每一個請求設置一個唯一的關聯ID，之后當我們從回調隊列收到一個響應的時候，我們將檢查這個屬性，基於此，便能將響應和請求關聯起來了。如果發現一個未知的關聯ID值，我們可以安全的銷毀消息，因為消息不屬於任何一個請求。

你可能會奇怪，為什么我們忽略掉未知關聯ID值得消息，而不是用錯誤來標記失敗？這是因為在服務器端可能存在爭用條件。盡管不太可能，但是RPC服務器可能在發送了響應消息而未發送消息確認的情況下出現故障，如果出現這樣的情況，在RPC服務器重啟之后將再次處理該請求。這就是為什么我們必須在客戶端優雅的捕獲重復的請求，並且RPC理論上應該是冪等的。

總結

                                 

我們的RPC將這樣工作：

• 當客戶端啟動時，它會創建一個匿名的獨占回調隊列。
• 對於一個RPC請求，客戶端通過兩個屬性發送一條消息：relayTo，設置回調隊列；correlationId，為每個請求設置一個唯一值。
• 消息將被發送到一個rpc_queue隊列。
• RPC工作線程（即，服務器）在該隊列上等待請求。當請求出現，他將處理請求並把結果發回給客戶端，使用的隊列是在replayTo中設置的。
• 客戶端在回調隊列上等待響應，當消息出現，它檢查關聯ID，如果匹配來自請求的關聯ID值，返回消息到該應用程序。

組合在一起

斐波那契任務：

private static int fib(int n)
{
    if (n == 0 || n == 1) return n;
    return fib(n - 1) + fib(n - 2);
}

我們定義斐波那契函數，它只采用正整數作為輸入。（別指望它能在大數值的情況下工作，而且這可能是最慢的一種遞歸實現）

RPC服務器RPCServer.cs中的代碼看起來是這樣的：

using System;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

class RPCServer
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            channel.QueueDeclare(queue: "rpc_queue",
                                 durable: false,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);
            channel.BasicQos(0, 1, false);
            var consumer = new QueueingBasicConsumer(channel);
            channel.BasicConsume(queue: "rpc_queue",
                                 noAck: false,
                                 consumer: consumer);
            Console.WriteLine(" [x] Awaiting RPC requests");

            while(true)
            {
                string response = null;
                var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                var body = ea.Body;
                var props = ea.BasicProperties;
                var replyProps = channel.CreateBasicProperties();
                replyProps.CorrelationId = props.CorrelationId;

                try
                {
                    var message = Encoding.UTF8.GetString(body);
                    int n = int.Parse(message);
                    Console.WriteLine(" [.] fib({0})", message);
                    response = fib(n).ToString();
                }
                catch(Exception e)
                {
                    Console.WriteLine(" [.] " + e.Message);
                    response = "";
                }
                finally
                {
                    var responseBytes = Encoding.UTF8.GetBytes(response);
                    channel.BasicPublish(exchange: "",
                                         routingKey: props.ReplyTo,
                                         basicProperties: replyProps,
                                         body: responseBytes);
                    channel.BasicAck(deliveryTag: ea.DeliveryTag,
                                     multiple: false);
                }
            }
        }
    }

    /// <summary>
    /// Assumes only valid positive integer input.
    /// Don't expect this one to work for big numbers,
    /// and it's probably the slowest recursive implementation possible.
    /// </summary>
    private static int fib(int n)
    {
        if(n == 0 || n == 1)
        {
            return n;
        }

        return fib(n - 1) + fib(n - 2);
    }
}

服務端代碼相當簡單：

• 通常情況下，我們都會以創建鏈接、信道和申明隊列作為開始。
• 我們可能希望運行不止一個服務器進程。為了將加載均勻分布到多個服務器，我們需要將prefetchCount設置為channel.basicQos。
• 我們使用basicConsume來訪問隊列。之后進入While循環，等待請求消息，完成工作，然后發回響應。

RPC客戶端RPCClient.cs中的代碼：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

class RPCClient
{
    private IConnection connection;
    private IModel channel;
    private string replyQueueName;
    private QueueingBasicConsumer consumer;

    public RPCClient()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        connection = factory.CreateConnection();
        channel = connection.CreateModel();
        replyQueueName = channel.QueueDeclare().QueueName;
        consumer = new QueueingBasicConsumer(channel);
        channel.BasicConsume(queue: replyQueueName,
                             noAck: true,
                             consumer: consumer);
    }

    public string Call(string message)
    {
        var corrId = Guid.NewGuid().ToString();
        var props = channel.CreateBasicProperties();
        props.ReplyTo = replyQueueName;
        props.CorrelationId = corrId;

        var messageBytes = Encoding.UTF8.GetBytes(message);
        channel.BasicPublish(exchange: "",
                             routingKey: "rpc_queue",
                             basicProperties: props,
                             body: messageBytes);

        while(true)
        {
            var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
            if(ea.BasicProperties.CorrelationId == corrId)
            {
                return Encoding.UTF8.GetString(ea.Body);
            }
        }
    }

    public void Close()
    {
        connection.Close();
    }
}

class RPC
{
    public static void Main()
    {
        var rpcClient = new RPCClient();

        Console.WriteLine(" [x] Requesting fib(30)");
        var response = rpcClient.Call("30");
        Console.WriteLine(" [.] Got '{0}'", response);

        rpcClient.Close();
    }
}

客戶端的代碼要稍微復雜一些：

• 創建一個鏈接、信道、為響應申明獨占的回調隊列。
• 訂閱回調隊列，以便接收RPC響應。
• call方法完成實際的RPC調用。
• 首先創建一個唯一的關聯Id並且保存它，while循環使用它去匹配合適的應答。
• 接下來，我們發布請求消息，使用了兩個屬性：replyTo和correlationId。
• 這時我們就可以坐等正確的響應到達了。
• While循環做的事情非常簡單，檢測每一個響應，如果correlactionId是我們需要的，就保存該響應。
• 最后，把響應返回給用戶。

構建客戶端請求：

RPCClient fibonacciRpc = new RPCClient();

System.out.println(" [x] Requesting fib(30)");
String response = fibonacciRpc.call("30");
System.out.println(" [.] Got '" + response + "'");

fibonacciRpc.close();

現在是時候來看看完整示例的源代碼了（包含基本的異常處理）。RPCClient.cs和RPCServer.cs。

編譯（參見教程一）：

$ csc /r:"RabbitMQ.Client.dll" RPCClient.cs
$ csc /r:"RabbitMQ.Client.dll" RPCServer.cs

現在RPC服務已經准備就緒，可以啟動服務了：

$ RPCServer.exe
 [x] Awaiting RPC requests

運行客戶端去請求斐波那契數列：

$ RPCClient.exe
 [x] Requesting fib(30)

這里介紹的設計並非RPC服務的唯一實現方式，但是它有一些重要的優勢：

• 如果RPC服務太慢，你可以通過運行另外一個實例來對其進行橫向擴展，試着在一個新的控制台里面運行另一個服務器。
• 在客戶端，RPC只要求發送和接收一條消息，沒有如同declareQueue的同步調用被要求。作為結果，RPC客戶端對於一個RPC請求，只需要一個網絡往返。

我們的代碼依然非常簡單，並沒有嘗試去解決一些復雜（但是重要）的問題，比如：

• 如果沒有運行中的服務器，客戶端將作何響應？
• 客戶端對於RPC是否可以有某種形式的超時？
• 如果服務器發生故障，引發異常，是否應當被轉發給客戶端？
• 在處理之前，避免無效的輸入數據，比如：檢查邊界、類型等。

 

如果你想嘗試，你可以找到有用的RabbitMQ管理插件去瀏覽隊列。

 

原文鏈接：http://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html