在教程二中，我们学习了如何使用工作队列在多个工作线程中分发耗时的任务。但如果我们需要去执行远程机器上的方法并且等待结果会怎么样呢？那又是另外一回事了。这种模式通常被称为远程过程调用（RPC）。
本教程中我们将使用RabbitMQ构建一个远程过程调用系统：一个客户端和一个可扩展的服务器。由于没有什么耗时的任务值得分发，我们将创建一个虚拟的RPC服务用于返回斐波那契数列。
为了阐释如何使用RPC服务我们将创建一个简单的客户端类。类中奖公开一个方法用于发送一个RPC请求，然后阻塞知道收到应答，方法名称叫做call:
1 var rpcClient = new RPCClient(); 2 3 Console.WriteLine(" [x] Requesting fib(30)"); 4 var response = rpcClient.Call("30"); 5 Console.WriteLine(" [.] Got ‘{0}‘", response); 6 7 rpcClient.Close();
RPC注记
尽管RPC在计算机技术中是一种非常常见的模式，但是它却饱受批判，问题发生在程序员不知道一个调用是本地的还是一个耗时的RPC。这样的混乱，导致不可预知的系统，并将不必要的复杂性调价到调试过程中。误用RPC将导致不可维护的混乱的代码，而不是简化软件。
铭记这些限制，考虑下面的建议：
应该在不能确定的时候避免使用RPC，如果可以的话，你可以使用异步管道，而不是类RPC的阻塞，结果被异步推送到下一个计算阶段。
一般来说，在RabbitMQ之上构建RPC非常的容易，客户端发送请求消息，服务返回应答消息。为了能够接收到应答的消息，我们需要在请求时指定一个回调队列地址：
1 var corrId = Guid.NewGuid().ToString(); 2 var props = channel.CreateBasicProperties(); 3 props.ReplyTo = replyQueueName; 4 props.CorrelationId = corrId; 5 6 var messageBytes = Encoding.UTF8.GetBytes(message); 7 channel.BasicPublish(exchange: "", 8 routingKey: "rpc_queue", 9 basicProperties: props, 10 body: messageBytes); 11 12 // ... 然后是从回调队列中读取消息的代码 ...
AMQP协议预定义了一个包含14个属性的属性集作用于消息之上，大多数都很少使用，除了下面这些：
在上面准备的方法中，我们建议为每一个RPC请求创建一个回调队列。这样相当低效，辛运的是有更好的方法，让我们为每一个客户端创建一个回调队列。
这样引出了一个新问题，当收到一个响应的时候，它无法清楚的知道响应属于哪一个请求。这就是correlationId派上用场的时候。我们将为每一个请求设置一个唯一的关联ID，之后当我们从回调队列收到一个响应的时候，我们将检查这个属性，基于此，便能将响应和请求关联起来了。如果发现一个未知的关联ID值，我们可以安全的销毁消息，因为消息不属于任何一个请求。
你可能会奇怪，为什么我们忽略掉未知关联ID值得消息，而不是用错误来标记失败？这是因为在服务器端可能存在争用条件。尽管不太可能，但是RPC服务器可能在发送了响应消息而未发送消息确认的情况下出现故障，如果出现这样的情况，在RPC服务器重启之后将再次处理该请求。这就是为什么我们必须在客户端优雅的捕获重复的请求，并且RPC理论上应该是幂等的。
我们的RPC将这样工作：
斐波那契任务：
1 private static int fib(int n) 2 { 3 if (n == 0 || n == 1) return n; 4 return fib(n - 1) + fib(n - 2); 5 }
我们定义斐波那契函数，它只采用正整数作为输入。（别指望它能在大数值的情况下工作，而且这可能是最慢的一种递归实现）
RPC服务器RPCServer.cs中的代码看起来是这样的：
1 using System; 2 using RabbitMQ.Client; 3 using RabbitMQ.Client.Events; 4 using System.Text; 5 6 class RPCServer 7 { 8 public static void Main() 9 { 10 var factory = new ConnectionFactory() { HostName = "localhost" }; 11 using(var connection = factory.CreateConnection()) 12 using(var channel = connection.CreateModel()) 13 { 14 channel.QueueDeclare(queue: "rpc_queue", 15 durable: false, 16 exclusive: false, 17 autoDelete: false, 18 arguments: null); 19 channel.BasicQos(0, 1, false); 20 var consumer = new QueueingBasicConsumer(channel); 21 channel.BasicConsume(queue: "rpc_queue", 22 noAck: false, 23 consumer: consumer); 24 Console.WriteLine(" [x] Awaiting RPC requests"); 25 26 while(true) 27 { 28 string response = null; 29 var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); 30 31 var body = ea.Body; 32 var props = ea.BasicProperties; 33 var replyProps = channel.CreateBasicProperties(); 34 replyProps.CorrelationId = props.CorrelationId; 35 36 try 37 { 38 var message = Encoding.UTF8.GetString(body); 39 int n = int.Parse(message); 40 Console.WriteLine(" [.] fib({0})", message); 41 response = fib(n).ToString(); 42 } 43 catch(Exception e) 44 { 45 Console.WriteLine(" [.] " + e.Message); 46 response = ""; 47 } 48 finally 49 { 50 var responseBytes = Encoding.UTF8.GetBytes(response); 51 channel.BasicPublish(exchange: "", 52 routingKey: props.ReplyTo, 53 basicProperties: replyProps, 54 body: responseBytes); 55 channel.BasicAck(deliveryTag: ea.DeliveryTag, 56 multiple: false); 57 } 58 } 59 } 60 } 61 62 /// <summary> 63 /// Assumes only valid positive integer input. 64 /// Don‘t expect this one to work for big numbers, 65 /// and it‘s probably the slowest recursive implementation possible. 66 /// </summary> 67 private static int fib(int n) 68 { 69 if(n == 0 || n == 1) 70 { 71 return n; 72 } 73 74 return fib(n - 1) + fib(n - 2); 75 } 76 }
服务端代码相当简单：
RPC客户端RPCClient.cs中的代码：
1 using System; 2 using System.Collections.Generic; 3 using System.Linq; 4 using System.Text; 5 using System.Threading.Tasks; 6 using RabbitMQ.Client; 7 using RabbitMQ.Client.Events; 8 9 class RPCClient 10 { 11 private IConnection connection; 12 private IModel channel; 13 private string replyQueueName; 14 private QueueingBasicConsumer consumer; 15 16 public RPCClient() 17 { 18 var factory = new ConnectionFactory() { HostName = "localhost" }; 19 connection = factory.CreateConnection(); 20 channel = connection.CreateModel(); 21 replyQueueName = channel.QueueDeclare().QueueName; 22 consumer = new QueueingBasicConsumer(channel); 23 channel.BasicConsume(queue: replyQueueName, 24 noAck: true, 25 consumer: consumer); 26 } 27 28 public string Call(string message) 29 { 30 var corrId = Guid.NewGuid().ToString(); 31 var props = channel.CreateBasicProperties(); 32 props.ReplyTo = replyQueueName; 33 props.CorrelationId = corrId; 34 35 var messageBytes = Encoding.UTF8.GetBytes(message); 36 channel.BasicPublish(exchange: "", 37 routingKey: "rpc_queue", 38 basicProperties: props, 39 body: messageBytes); 40 41 while(true) 42 { 43 var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); 44 if(ea.BasicProperties.CorrelationId == corrId) 45 { 46 return Encoding.UTF8.GetString(ea.Body); 47 } 48 } 49 } 50 51 public void Close() 52 { 53 connection.Close(); 54 } 55 } 56 57 class RPC 58 { 59 public static void Main() 60 { 61 var rpcClient = new RPCClient(); 62 63 Console.WriteLine(" [x] Requesting fib(30)"); 64 var response = rpcClient.Call("30"); 65 Console.WriteLine(" [.] Got ‘{0}‘", response); 66 67 rpcClient.Close(); 68 } 69 }
客户端的代码要稍微复杂一些：
构建客户端请求：
1 RPCClient fibonacciRpc = new RPCClient(); 2 3 System.out.println(" [x] Requesting fib(30)"); 4 String response = fibonacciRpc.call("30"); 5 System.out.println(" [.] Got ‘" + response + "‘"); 6 7 fibonacciRpc.close();
现在是时候来看看完整示例的源代码了（包含基本的异常处理）。RPCClient.cs和RPCServer.cs。
编译（参见教程一）：
1 $ csc /r:"RabbitMQ.Client.dll" RPCClient.cs 2 $ csc /r:"RabbitMQ.Client.dll" RPCServer.cs
现在RPC服务已经准备就绪，可以启动服务了：
1 $ RPCServer.exe 2 [x] Awaiting RPC requests
运行客户端去请求斐波那契数列：
1 $ RPCClient.exe 2 [x] Requesting fib(30)
这里介绍的设计并非RPC服务的唯一实现方式，但是它有一些重要的优势：
我们的代码依然非常简单，并没有尝试去解决一些复杂（但是重要）的问题，比如：
如果你想尝试，你可以找到有用的RabbitMQ管理插件去浏览队列。
原文链接：http://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html
原文：http://www.cnblogs.com/chen108/p/4972875.html