远程过程调用（RPC）
(使用pika 0.9.5 python客户端)
在第二篇说明里，我们学习了如何在多个worker中使用Work Queues分配耗时的任务。
但是如果我们想在远程机器上运行程序，并得到结果？
那么，这儿有一个故事。它就被称作远程过程调用或者RPC。
在这篇说明里，我们的目标是使用RabbitMQ建立一个RPC系统：
一个客户端和一个可扩展的RPC服务器。当然我们没有任何需要耗时的任务需要分发。
我们的目标是建立一个模拟的RPC服务，它只用来计算斐波那契数列并返回。
客户端：
为了举例说明什么是RPC服务，我们创建了一个简单的客户端类。
它是这样的，一个叫做call的方法发送了RPC请求，并且等待结果的返回：
1 fibonacci_rpc = FibonacciRpcClient() 2 result = fibonacci_rpc.call(4) 3 print "fib(4) is %r" % (result,)
关于RPC：
尽管RPC在计算中很普遍，但是它却经常受到批评。当一个程序员没有意识到，一个函数调用
是本地，或者是一个缓慢的RPC调用的时候，问题便出现了。这种结果就导致了不可预知的系统，
并且增加了代码调试的复杂性。相对于简单的软件，滥用RPC会导致像意大利面一样缠绕不清的
代码，这些代码很难维护。
Bearing that in the mind（该怎么翻译？？？)，考虑下面的建议：
> 明确的标明那些函数调用是本地的，哪些是远程调用。
> 文档化你的系统。让各个组件之间的依赖关系保持清晰。
> 处理错误。当RPC服务器挂掉了你的客户端会怎样处理？
不要太怀疑RPC。如果可以，你可以使用异步的管道，代替RPC这样的阻塞调用。
返回的结果异步的PUSH到下一个计算步骤。
回调队列
一般来说，使用RabbitMQ做RPC很简单。一个客户端发送请求消息，服务器返回结果。
为了能收到返回的结果，客户端需要在请求的信息中，包含一个"回调"队列。
让我们试一试：
1 result = channel.queue_declare(exclusive=True) 2 callback_queue = result.method.queue 3 4 channel.basic_publish(exchange='', 5 routing_key='rpc_queue', 6 properties=pika.BasicProperties( 7 reply_to = callback_queue, 8 ), 9 body=request) 10 11 # ...下面的代码,读取从callback_queue中读取返回的消息...
消息的属性
AMQP协议为一个消息定义了14个属性，大部分都很少使用，但是下面的比较特殊：
> delivery_mode：标明了一个消息是持久的（2）或者是临时的（其他数字）。
你可能还记得在第二篇说明中提到了这个属性。
> content_type：用来描述MIME类型的编码。举个例子来说，就像我们经常使用的JSON格式，
我们习惯性的用下面的编码来描述它：
application/json
> reply_to：一般来说，它只是回调队列的名称
> correlation_id：用来关联RPC返回和请求之间的ID。
关联ID
在上面的方法中，我们建议为每个RPC请求建立一个回调队列。但是这样效率太低了，
幸运的是，有更好的方法：为每个客户端建立一个回调队列。
但是这样带来一个新的问题，在回调队列中我们无法区分一个返回结果属于哪个特定的请求。
这就是使用correlation_id的原因。在每个请求中我们设置correlation_id为唯一的值。
然后，在回调队列中收到一个消息，我们会查看correlation_id，基于它我们就可以让
请求和返回之间匹配。如果我们发现了一个未知的correlation_id，出于安全，我们会丢弃它。
因为它不属于我们的请求。
你可能会问，为什么会在回调队列中忽略未知的消息，而不是抛出一个错误？
因为可能在服务端会发生一些极端的情况。
当服务端发送一个返回结果给客户端的时候，但是在它发送确认消息给请求之前，它有可能会挂掉，尽管不太可能发生。
如果发生了，服务端在重启之后，会重新处理在它挂掉之前收到的且未处理的请求。
这就是为什么在客户端我们需要以优雅的方式处理重复的返回结果，这样RPC就会更完美。
总揽
我们的RPC服务的工作方式：
> 当客户端启动，它会创建一个匿名的而且*排他*的回调队列。
> 客户端为每个请求设置两个属性：reply_to，回调队列的名称和correlation_id，每个请求的唯一ID。
> 请求被发送到一个叫做 rpc_queue 的队列。
> RPC worker(就是服务器)等待在 rpc_queue 队列上，当出现一个消息，服务器处理任务，
然后使用 reply_to 中标明的回调队列，发送返回结果给客户端，
> 客户端在回调队列上等待。当消息出现，客户端检查correaltion_id的值。
如果这个值和请求中的correlation_id值匹配，就返回结果给应用程序。
放在一起来看：
rpc_server.py的代码：
1 #!/usr/bin/env python 2 import pika 3 4 connection = pika.BlockingConnection(pika.ConnectionParameters( 5 host='localhost')) 6 7 channel = connection.channel() 8 9 channel.queue_declare(queue='rpc_queue') 10 11 def fib(n): 12 if n == 0: 13 return 0 14 elif n == 1: 15 return 1 16 else: 17 return fib(n-1) + fib(n-2) 18 19 def on_request(ch, method, props, body): 20 n = int(body) 21 22 print " [.] fib(%s)" % (n,) 23 response = fib(n) 24 25 ch.basic_publish(exchange='', 26 routing_key=props.reply_to, 27 properties=pika.BasicProperties(correlation_id = \ 28 props.correlation_id), 29 body=str(response)) 30 ch.basic_ack(delivery_tag = method.delivery_tag) 31 32 channel.basic_qos(prefetch_count=1) 33 channel.basic_consume(on_request, queue='rpc_queue') 34 35 print " [x] Awaiting RPC requests" 36 channel.start_consuming()
服务端的代码比较简单：
> 建立到RabbitMQ服务器的连接，并且声明一个叫rpc_queue的队列。
> 定义计算波那契数列的函数，函数只计算正整数。
(不要指望这个函数能计算很大的数，它可能是一个比较缓慢的回调的实现。)
> 我们为basic_consume指定一个回调函数，这个回调函数是RPC服务器的核心。
当收到请求时，这个函数就会被调用。它处理工作后发送结果。
> 我们需要运行不止一个RPC服务器进程，当我们需要在多个服务器之间公平的展开任务处理时。
就需要设置prefetch_count属性。
rpc_client.py的代码：
1 #!/usr/bin/env python 2 import pika 3 import uuid 4 5 class FibonacciRpcClient(object): 6 def __init__(self): 7 self.connection = pika.BlockingConnection(pika.ConnectionParameters( 8 host='localhost')) 9 10 self.channel = self.connection.channel() 11 12 result = self.channel.queue_declare(exclusive=True) 13 self.callback_queue = result.method.queue 14 15 self.channel.basic_consume(self.on_response, no_ack=True, 16 queue=self.callback_queue) 17 18 def on_response(self, ch, method, props, body): 19 if self.corr_id == props.correlation_id: 20 self.response = body 21 22 def call(self, n): 23 self.response = None 24 self.corr_id = str(uuid.uuid4()) 25 self.channel.basic_publish(exchange='', 26 routing_key='rpc_queue', 27 properties=pika.BasicProperties( 28 reply_to = self.callback_queue, 29 correlation_id = self.corr_id, 30 ), 31 body=str(n)) 32 while self.response is None: 33 self.connection.process_data_events() 34 return int(self.response) 35 36 fibonacci_rpc = FibonacciRpcClient() 37 38 print " [x] Requesting fib(30)" 39 response = fibonacci_rpc.call(30) 40 print " [.] Got %r" % (response,)
客户端的代码有些复杂：
> 建立到RabbitMQ服务器的连接，建立一个channel，并且为了收到返回结果，我们声明一个排他的"回调"队列。
> 我们"订阅"这个回调队列，这样就能收到RPC的返回结果。
> 当结果返回时，on_response方法只做一些很简单的任务，它检查每个返回消息中的correlation_id是不是我们
想要的那个ID。如果是，这个方法就会保存返回结果到self.response中，并且中断consuming循环。
> 接下来，定义了主要的call方法，它发出RPC请求。
> 在call方法中，首先产生一个唯一的correlation_id并且保存。
on_response方法会使用这个唯一的ID来捕捉正确的返回结果。
> 然后，我们发送一个请求，这个请求有两个属性：reply_to和correlation_id。
> 这个时候我们会等到结果返回。
> 最终返回结果给用户。
我们的RPC服务已经就绪，可以运行它了：
1 $ python rpc_server.py 2 [x] Awaiting RPC requests
运行客户端，发送请求：
1 $ python rpc_client.py 2 [x] Requesting fib(30)
上面的设计中，也许不是全部的RPC服务的实现过程。但它的确有一些高级并且重要的特新：
> 如果RPC服务器很慢，只要运行另外一个RPC服务器就能扩展它。可以尝试在控制台运行第二个RPC服务器。
> 在客户端，发送RPC请求并且只接收一个消息。不需要像queue_declare这样的同步方法(这样翻译对否？)。
在一个RPC请求中，客户端只需要做一次网络发送和请求的过程。
我们的代码依然相当简单，并且不准备解决一些很复杂但是很重要的问题，就像下面这样：
> 如果服务端没有运行，客户端应该怎样处理？
> 在一个RPC请求过程中，客户端怎样处理超时？
> 如果RPC服务器出现异常，是否需要通知客户端？
> 在处理消息之前，检查消息的边界，以此来防止有异常的消息进入。