在第二个教程中，我们了解到如何在多个worker中使用Work Queues分发费时的任务。
但是，如果我们需要在远程运行一个函数并且等待结果该怎么办呢？这个时候，我们需要另外一个模式了。这种模式通常被叫做Remote Procedure Call 或者RPC.
在这个教程中，我们将使用RabbitMQ来建立一个RPC系统：a client和a scalable RPC server.
Client interface
为了说明RPC服务怎样被使用，我们将创建一个简单的Client class(客户端类)。它会暴露一个发送RPC请求的名叫Call的方法并且会阻塞到接收到answer.
代码语言：javascript复制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(回调队列地址).
代码语言：javascript复制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:
代码语言：javascript复制private static int fib(int n)
{
if (n == 0 || n == 1) return n;
return fib(n - 1) fib(n - 2);
}RPCServer.cs
代码语言：javascript复制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
代码语言：javascript复制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
