遠程調用(RPC)
ps: 使用pika
以前的筆記中买决，已經知道如何使用工作隊列鸟整，在多個工作者Worker中分發(fā)耗時任務采蚀。
但是有時候我們可能遇到這樣的情況澎嚣，我們將一個函數(shù)運行在遠程計算機上，然后另一個計算機等待從remote Computer上獲取運行的結果淋淀，該怎么辦遥昧？這種模式通常叫做遠程過程調用(Remote Procedure Call)就是RPC。
客戶端接口
為了展示PRC服務的使用朵纷，我們創(chuàng)建一個簡單的客戶頓類炭臭。它暴露出一個名為call()的方法用來發(fā)送一個RPC請求，并且在收到回應前保持阻塞袍辞。
fibonacci_rpc = FibonacciPrcClient()
result = fibonacci_rpc .call(4)
print " fib(4) is %r" % (result, )
回調隊列
RabbbitMQ實現(xiàn)RPC是很容易的鞋仍，一個客戶端發(fā)送來請求信息，服務端將應用到一個回復信息中搅吁。為了接受到回復信息威创，客戶端需要在發(fā)送請求的時候同時發(fā)送一個回調隊列(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)
消息屬性
AMQP協(xié)議給消息定義了一系列的屬性，常用的屬性有：
- delivery_mode（投遞模式）：將消息標記為持久的（值為2）或暫存的（除了2之外的其他任何值）肚豺。第二篇教程里接觸過這個屬性溃斋，記得吧？
- content_type（內容類型）:用來描述編碼的mime-type吸申。例如在實際使用中常常使用application/json來描述JOSN編碼類型梗劫。
- reply_to（回復目標）：通常用來命名回調隊列。
- correlation_id（關聯(lián)標識）：用來將RPC的響應和請求關聯(lián)起來
關聯(lián)標識
在每一個RPC請求都創(chuàng)建一個回調隊列截碴，這樣如果請求數(shù)量多大導致系統(tǒng)內隊列過多梳侨，造成性能下降，所以更好的方法就是為每一個客戶端只創(chuàng)建一個獨立回調隊列日丹。
但是有一個問題就是猫妙，一個客戶端可能發(fā)送很多個請求，服務端處理完后怎么知道把相應發(fā)送給那個請求呢聚凹？correlation_id可以解決這個問題割坠，相當于就是每一個請求會有有一個獨一無二的值標識，當相應時妒牙，會匹配這個獨一無二的值彼哼，將相應發(fā)送到指定的請求。
總結
如何工作
- 客戶端啟動的時候湘今，會創(chuàng)建一個回調隊列(可以使用匿名隊列)敢朱。
- 在發(fā)送RPC請求時，客戶端發(fā)送帶有兩個屬性的消息摩瞎，一個設置回調隊列的
reply_to屬性拴签，另一個設置唯一值correlation_id屬性. - 將請求發(fā)送到一個
rpc_queue隊列中。 - RPC工作者(服務器)等待請求發(fā)送到這個隊列中旗们，然后獲取請求并執(zhí)行蚓哩，將執(zhí)行的結果發(fā)送到
reply_to指定的隊列中。 - 客戶端等待回調隊列中的數(shù)據上渴，并檢查
correlation_id屬性岸梨，如果屬性匹配就將數(shù)據取走。
代碼
rpc_server.py
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()
解釋
- 我們先創(chuàng)建連接稠氮，聲明隊列曹阔。
- 聲明fibonacci函數(shù)，在輸入一個合法的值返回一個結果隔披，但是遞歸不能太深赃份。
- 為 basic_consume 聲明了一個回調函數(shù)，這也是RPC設計的核心奢米，它主要是執(zhí)行任務并產生相應抓韩。
rpc_client.py
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,)
解釋
- 創(chuàng)建連接纠永、并聲明獨享的回調隊列
- 訂閱這個回調隊列，便于接受RPC相應园蝠。
- “on_response”回調函數(shù)對每一個響應執(zhí)行一個非常簡單的操作，檢查每一個響應消息的correlation_id屬性是否與我們期待的一致痢士，如果一致彪薛，將響應結果賦給self.response，然后跳出consuming循環(huán)怠蹂。
- 我們定義我們的主要方法 call 方法善延。它執(zhí)行真正的RPC請求。
- 在這個方法中城侧，首先我們生成一個唯一的 correlation_id 值并且保存起來易遣，'on_response'回調函數(shù)會用它來獲取符合要求的響應。
- 接下來嫌佑，我們將帶有 reply_to 和 correlation_id 屬性的消息發(fā)布出去豆茫。
- 現(xiàn)在我們可以坐下來，等待正確的響應到來屋摇。
- 最后揩魂，我們將響應返回給用戶。
