• 6 RPC

  Remote procedure call implementation

  

  Python | Java | Ruby | PHP| C#

Remote procedure call (RPC)

(using the .NET client)

在第二個教程second tutorial 中我們已經瞭解到了工作佇列如何將耗時任務分配給多個workers。

但是假如我們需要在遠端機器上面執行一個函式並且等待結果返回呢？這通常叫做RPC，即遠端過程呼叫。

這裡我們將用RabbitMQ構造一個RPC系統，客戶端請求呼叫服務端的計算斐波納契數列值得一個函式，並等待計算結果。

Client interface（客戶端呼叫介面）

首先看一下客戶端介面，我們定義一個RPC呼叫類，其中提供了一個叫做Call的介面，這個介面內部所做的事情就是將呼叫服務端計算斐波那契數列的請求（包含引數）傳送到指定的訊息佇列，然後再另一個臨時佇列阻塞等待服務端將計算結果放入到這個臨時佇列。

var rpcClient = new RPCClient();

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

rpcClient.Close
 
();

Callback queue（回撥佇列）

下面程式碼是Call介面內部實現的一部分，為了等待服務端RPC呼叫的結果，我們需要告訴服務端將計算結果放到哪個佇列中，這裡props引數就已經制定了計算結果的存放佇列名稱，同時還附上了每個RPC請求的ID，方便讀取response的時候能夠知道對應於哪個請求：

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

var messageBytes = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "rpc_queue", props, messageBytes);

// ... then code to read a response message from the callback_queue ...

Correlation Id（RPC請求ID）

很顯然我們不可能為每個RPC呼叫都建立一個存放呼叫結果的回撥佇列，我們可以為每個client端都建立一個。

至於每個RPC請求發出去之後，收到迴應時如何知道這個response是對應於哪個RPC請求，就需要用到 correlationId 屬性. 

這個ID值可以有多重方法生成，比如客戶端IP+計數值，或者一個唯一的GUID等都可以。

Summary（總結）

RPC呼叫工作流程:

• 客戶端啟動時候建立一個匿名獨佔的回撥佇列。
• 進行RPC呼叫時，傳送帶有兩個屬性的訊息（RPC請求）到指定佇列，這兩個屬性是指定回撥佇列名稱的 replyTo 屬性，以及唯一標識一個RPC請求的 correlationId 屬性。
• 將RPC請求傳送到約定好的 rpc_queue 佇列。
• 服務端的RPC worker收到RPC請求之後開始呼叫函式，執行完成之後將執行結果放到 replyTo 指定的回撥佇列中。
• 客戶端在回撥佇列上等待呼叫結果，當收到訊息之後檢視 correlationId 屬性時候是剛才的RPC請求的response，如果是就返回。

Putting it all together（程式碼總覽）

斐波那契數列計算函式:

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

class RPCServer
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare("rpc_queue", false, false, false, null);
                channel.BasicQos(0, 1, false);
                var consumer = new QueueingBasicConsumer(channel);
                channel.BasicConsume("rpc_queue", false, 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("", props.ReplyTo, replyProps, responseBytes);
                        channel.BasicAck(ea.DeliveryTag, false);
                    }
                }
            }
        }
    }

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

The server code is rather straightforward:

• 和往常一樣建立連線，channel，以及申明佇列。
• 你可能想要執行多個server程序，為了讓請求均勻地負載到多個servers上面，我們需要呼叫 channel.basicQos. 告訴RabbitMQ不要將超過一個的訊息同時分配給同一個worker，詳見：http://blog.csdn.net/jiyiqinlovexx/article/details/38946955
• 呼叫basicConsume 訪問佇列. 
• 然後進入迴圈，等待訊息，執行函式呼叫，返回response。.

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();
        consumer = new QueueingBasicConsumer(channel);
        channel.BasicConsume(replyQueueName, true, 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("", "rpc_queue", props, 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();
    }
}

The client code is slightly more involved:

• 建立connection和channel，申明一個獨佔的臨時回撥佇列。
• 訂閱回撥佇列，以便能夠接收RPC請求的response。
• 實現Call方法，用於真正傳送RPC請求
• 首先生成唯一的 correlationId 值儲存下來，while迴圈中將會用這個值來匹配response。
• 建立RPC請求訊息，包含兩個屬性: replyTo and correlationId.
• 發出RPC請求，等待迴應.
• while迴圈對於每一個response都會檢查 thecorrelationId 是否匹配，如果匹配就儲存下來這個response.
• 最後返回response給使用者.

測試程式，開始進行RPC呼叫:

RPCClient fibonacciRpc = new RPCClient();

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

fibonacciRpc.close();

Now is a good time to take a look at our full example source code (which includes basic exception handling) for RPCClient.cs and RPCServer.cs.

Compile as usual (see tutorial one):

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

Our RPC service is now ready. We can start the server:

$ RPCServer.exe
 [x] Awaiting RPC requests

To request a fibonacci number run the client:

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

The design presented here is not the only possible implementation of a RPC service, but it has some important advantages:

• If the RPC server is too slow, you can scale up by just running another one. Try running a second RPCServer in a new console.
• On the client side, the RPC requires sending and receiving only one message. No synchronous calls like queueDeclare are required. As a result the RPC client needs only one network round trip for a single RPC request.

Our code is still pretty simplistic and doesn't try to solve more complex (but important) problems, like:

• How should the client react if there are no servers running?
• Should a client have some kind of timeout for the RPC?
• If the server malfunctions and raises an exception, should it be forwarded to the client?
• Protecting against invalid incoming messages (eg checking bounds, type) before processing.

If you want to experiment, you may find the rabbitmq-management plugin useful for viewing the queues.