我们介绍了如何使用工作队列（work queue）在多个工作者（woker）中间分发耗时的任务。
可是如果我们需要将一个函数运行在远程计算机上并且等待从那儿获取结果时，该怎么办呢？这就是另外的故事了。这种模式通常被称为远程过程调用（Remote Procedure Call）或者RPC。
这篇教程中，我们会使用RabbitMQ来构建一个RPC系统：包含一个客户端和一个RPC服务器。现在的情况是，我们没有一个值得被分发的足够耗时的任务，所以接下来，我们会创建一个模拟RPC服务来返回斐波那契数列。
为了展示RPC服务如何使用，我们创建了一个简单的客户端类。它会暴露出一个名为“call”的方法用来发送一个RPC请求，并且在收到回应前保持阻塞。
| fibonacci_rpc = FibonacciRpcClient() |
| result = fibonacci_rpc.call(4) |
| print "fib(4) is %r" % (result,) |
尽管RPC在计算领域是一个常用模式，但它也经常被诟病。当一个问题被抛出的时候，程序员往往意识不到这到底是由本地调用还是由较慢的RPC调用引起的。同样的困惑还来自于系统的不可预测性和给调试工作带来的不必要的复杂性。跟软件精简不同的是，滥用RPC会导致不可维护的面条代码.
考虑到这一点，牢记以下建议：
确保能够明确的搞清楚哪个函数是本地调用的，哪个函数是远程调用的。给你的系统编写文档。保持各个组件间的依赖明确。处理错误案例。明了客户端改如何处理RPC服务器的宕机和长时间无响应情况。
当对避免使用RPC有疑问的时候。如果可以的话，你应该尽量使用异步管道来代替RPC类的阻塞。结果被异步地推送到下一个计算场景。
一般来说通过RabbitMQ来实现RPC是很容易的。一个客户端发送请求信息，服务器端将其应用到一个回复信息中。为了接收到回复信息，客户端需要在发送请求的时候同时发送一个回调队列（callback queue）的地址。我们试试看：
| result = channel.queue_declare(exclusive=True) |
| callback_queue = result.method.queue |
| channel.basic_publish(exchange='', |
| routing_key='rpc_queue', |
| properties=pika.BasicProperties( |
| reply_to = callback_queue, |
| ), |
| body=request) |
| # ... and some code to read a response message from the callback_queue ... |
AMQP协议给消息预定义了一系列的14个属性。大多数属性很少会用到，除了以下几个：
上边介绍的方法中，我们建议给每一个RPC请求新建一个回调队列。这不是一个高效的做法，幸好这儿有一个更好的办法 —— 我们可以为每个客户端只建立一个独立的回调队列。
这就带来一个新问题，当此队列接收到一个响应的时候它无法辨别出这个响应是属于哪个请求的。correlation_id 就是为了解决这个问题而来的。我们给每个请求设置一个独一无二的值。稍后，当我们从回调队列中接收到一个消息的时候，我们就可以查看这条属性从而将响应和请求匹配起来。如果我们接手到的消息的correlation_id是未知的，那就直接销毁掉它，因为它不属于我们的任何一条请求。
你也许会问，为什么我们接收到未知消息的时候不抛出一个错误，而是要将它忽略掉？这是为了解决服务器端有可能发生的竞争情况。尽管可能性不大，但RPC服务器还是有可能在已将应答发送给我们但还未将确认消息发送给请求的情况下死掉。如果这种情况发生，RPC在重启后会重新处理请求。这就是为什么我们必须在客户端优雅的处理重复响应，同时RPC也需要尽可能保持幂等性。
我们的RPC如此工作:
rpc_server.py代码：
| #!/usr/bin/env python |
| import pika |
| connection = pika.BlockingConnection(pika.ConnectionParameters( |
| host='localhost')) |
| channel = connection.channel() |
| channel.queue_declare(queue='rpc_queue') |
| def fib(n): |
| if n == 0: |
| return 0 |
| elif n == 1: |
| return 1 |
| else: |
| return fib(n-1) + fib(n-2) |
| def on_request(ch, method, props, body): |
| n = int(body) |
| print " [.] fib(%s)" % (n,) |
| response = fib(n) |
| ch.basic_publish(exchange='', |
| routing_key=props.reply_to, |
| properties=pika.BasicProperties(correlation_id = \ |
| props.correlation_id), |
| body=str(response)) |
| ch.basic_ack(delivery_tag = method.delivery_tag) |
| channel.basic_qos(prefetch_count=1) |
| channel.basic_consume(on_request, queue='rpc_queue') |
| print " [x] Awaiting RPC requests" |
| channel.start_consuming() |
rpc_client.py 代码:
| #!/usr/bin/env python |
| import pika |
| import uuid |
| class FibonacciRpcClient(object): |
| def __init__(self): |
| self.connection = pika.BlockingConnection(pika.ConnectionParameters( |
| host='localhost')) |
| self.channel = self.connection.channel() |
| result = self.channel.queue_declare(exclusive=True) |
| self.callback_queue = result.method.queue |
| self.channel.basic_consume(self.on_response, no_ack=True, |
| queue=self.callback_queue) |
| def on_response(self, ch, method, props, body): |
| if self.corr_id == props.correlation_id: |
| self.response = body |
| def call(self, n): |
| self.response = None |
| self.corr_id = str(uuid.uuid4()) |
| self.channel.basic_publish(exchange='', |
| routing_key='rpc_queue', |
| properties=pika.BasicProperties( |
| reply_to = self.callback_queue, |
| correlation_id = self.corr_id, |
| ), |
| body=str(n)) |
| while self.response is None: |
| self.connection.process_data_events() |
| return int(self.response) |
| fibonacci_rpc = FibonacciRpcClient() |
| print " [x] Requesting fib(30)" |
| response = fibonacci_rpc.call(30) |
| print " [.] Got %r" % (response,) |
我们的RPC服务已经准备就绪了，现在启动服务器端：
运行客户端，请求一个fibonacci队列。
| $ python rpc_client.py |
| [x] Requesting fib(30) |
此处呈现的设计并不是实现RPC服务的唯一方式，但是他有一些重要的优势：
我们的代码依旧非常简单，而且没有试图去解决一些复杂（但是重要）的问题，如：
（完整的rpc_client.py 和 rpc_server.py代码)
