RabbitMQ学习之RPC(6)

 在第二个教程中，我们了解到如何在多个worker中使用Work Queues分发费时的任务。

但是，如果我们需要在远程运行一个函数并且等待结果该怎么办呢？这个时候，我们需要另外一个模式了。这种模式通常被叫做Remote Procedure Call 或者RPC.

在这个教程中，我们将使用RabbitMQ来建立一个RPC系统：a client和a scalable RPC server.

Client interface

 为了说明RPC服务怎样被使用，我们将创建一个简单的Client class(客户端类)。它会暴露一个发送RPC请求的名叫Call的方法并且会阻塞到接收到answer. 

var rpcClient = new RPCClient();

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

rpcClient.Close();

Callback queue

一般来说，通过RabbitMQ来做RPC是简单的。客户端(client)发送request message并且服务端(server)返回response message. 为了接收到response，我们需要在request上发送一个callback queue address(回调队列地址). 

var props = channel.CreateBasicProperties();
props.ReplyTo = replyQueueName;  //设置callback queue name
var messageBytes = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
                     routingKey: "rpc_queue",
                     basicProperties: props,
                     body: messageBytes);
// ... then code to read a response message from the callback_queue ...

Message properties

AMQP协议在message上预定义了14个属性的集合。 大部分属性很少使用，下面的使用比较多：

• Persistent : 使message持久化

• DeliveryMode : 那些熟悉这个协议的可能会使用这个属性而不是Persistent.来做持久化。

• ContentType : 用来描述编码类型。例如，经常使用的JSON编码，通常设置属性为：application/json

• ReplyTo : 用来命名callback queue(回调队列)

• CorrelationId : 用来关联RPC Response 和request

Correlation Id

在之前我们讲的方法中，我们建议为每一个RPC request建立一个callback queue. 那样很没有效率，幸运的，还有一种更好的方法：我们为每个client创建单独的一个callback queue.

这个时候我们需要CorrelationId属性来关联response和request. 每个request都有唯一的correlationId. 当我们在队列中收到一个message，我们看下这个属性，并且根据它我们来匹配response和request. 如果我们看到一个不知道的CorrelationId值，我们会安全的丢掉这个message. 它不属于我们的requests. 

你可能会问，为什么我们忽视callback queue中不知道的message（unknow messages），而不是报错呢?那是服务端有 竞态资源 的可能性。尽管不太可能，但它是可能的，RPC服务器在发送给我们answer之后，但还没有发送an acknowledgement message之前死掉了。如果这种情况发生了，重启的RPC服务器将会再处理这个request. 那就是客户端为什么要优雅的处理两次responses. (可以对比第二个教程，会在接收端确认，如果接收端没有确认，之后队列会再次发送request，服务端需要再次处理)

Summary(总结)

我们的RPC像图中这样工作：

• 当一个client启动时，它创建一个匿名的专用的callback queue.
• 对于一个RPC request，client将会发送带有两个属性的message。ReplyTo,用来设置callback queue；并且CorrelationId,用来为每个request设置唯一的值。
• Request会被发送到rpc_queue.
• RPC worker（这里就是server）将会等待接收rpc_queue队列里的requests。当request出现时，它会做这个job,并且发送一个带结果的message给client,使用被ReplyTo属性命名的队列。
• Client会等待callback queue的数据。当一个message出现时，它会检查CorrealtionId属性。如果它匹配request里的这个值(CorrealationId)，将会返回response到这个应用(client)

代码

The Fibonacci task:

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

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,  //声明queue
              exclusive: false, autoDelete: false, arguments: null);
            channel.BasicQos(0, 1, false); //公平调度策略
            var consumer = new EventingBasicConsumer(channel);
            channel.BasicConsume(queue: "rpc_queue",  //接收消息
              autoAck: false, consumer: consumer);
            Console.WriteLine(" [x] Awaiting RPC requests");

            consumer.Received += (model, ea) =>
            {
                string response = null;

                var body = ea.Body;
                var props = ea.BasicProperties;
                var replyProps = channel.CreateBasicProperties();
                replyProps.CorrelationId = props.CorrelationId;//设置返回的CorrealationId

                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,//发送响应到callback queue
                      basicProperties: replyProps, body: responseBytes);
                    channel.BasicAck(deliveryTag: ea.DeliveryTag,
                      multiple: false);
                }
            };

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }

    /// 
    /// 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.
    /// 
    private static int fib(int n)
    {
        if (n == 0 || n == 1)
        {
            return n;
        }

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

RPCClient.cs

using System;using System.Collections.Concurrent;using System.Text;using RabbitMQ.Client;using RabbitMQ.Client.Events;
public class RpcClient
{
    private readonly IConnection connection;
    private readonly IModel channel;
    private readonly string replyQueueName;
    private readonly EventingBasicConsumer consumer;
    private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>();
    private readonly IBasicProperties props;
public RpcClient()
{
        var factory = new ConnectionFactory() { HostName = "localhost" };

        connection = factory.CreateConnection();
        channel = connection.CreateModel();
        replyQueueName = channel.QueueDeclare().QueueName;
        consumer = new EventingBasicConsumer(channel);

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

        consumer.Received += (model, ea) =>
        {
            var body = ea.Body;
            var response = Encoding.UTF8.GetString(body);
            if (ea.BasicProperties.CorrelationId == correlationId)
            {
                respQueue.Add(response);
            }
        };
    }

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

        channel.BasicConsume(
            consumer: consumer,
            queue: replyQueueName,
            autoAck: true);

        return respQueue.Take(); ;
    }

    public void Close()
    {
        connection.Close();
    }
}
public 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();
    }
}

 参考网址：

https://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html