RabbitMQ学习之RPC(6)

时间 2019-11-11

原文原文链接

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

可是，若是咱们须要在远程运行一个函数而且等待结果该怎么办呢？这个时候，咱们须要另一个模式了。这种模式一般被叫作Remote Procedure Call 或者RPC.json

在这个教程中，咱们将使用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(回调队列地址). app

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持久化ui
DeliveryMode : 那些熟悉这个协议的可能会使用这个属性而不是Persistent.来作持久化。this
ContentType : 用来描述编码类型。例如，常常使用的JSON编码，一般设置属性为：application/json编码
ReplyTo : 用来命名callback queue(回调队列)spa
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