Python rabbitmq的使用（七）

上一遍演示了遠程結果返回的示例，可是有一個沒有提到，就是correlation id，這個是個什麼東東呢？

假設有多個計算節點，控制中心開啓多個線程，往這些計算節點發送數字，要求計算結果並返回，可是控制中心只開啓了一個隊列，全部線程都是從這個隊列裏獲取消息，每一個線程如何肯定收到的消息就是該線程對應的呢？這個就是correlation id的用處了。correlation翻譯成中文就是相互關聯，也表達了這個意思。

correlation id運行原理：控制中心發送計算請求時設置correlation id，然後計算節點將計算結果，連同接收到的correlation id一塊兒返回，這樣控制中心就能經過correlation id來標識請求。其實correlation id也能夠理解爲請求的惟一標識碼。

示例內容：控制中心開啓多個線程，每一個線程都發起一次計算請求，經過correlation id，每一個線程都能準確收到相應的計算結果。

compute.py代碼分析

和上面一篇相比，只需修改一個地方：將計算結果發送回控制中心時，增長參數correlation_id的設定，該參數的值實際上是從控制中心發送過來的，這裏只是再次發送回去。代碼以下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

#!/usr/bin/env python #coding=utf8 import pika   #鏈接rabbitmq服務器 connection = pika.BlockingConnection(pika.ConnectionParameters(         host='localhost')) channel = connection.channel()   #定義隊列 channel.queue_declare(queue='compute_queue') print ' [*] Waiting for n'   #將n值加1 def increase(n):     return n + 1   #定義接收到消息的處理方法 def request(ch, method, props, body):     print " [.] increase(%s)"  % (body,)       response = increase(int(body))       #將計算結果發送回控制中心，增長correlation_id的設定     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(request, queue='compute_queue')   channel.start_consuming()

center.py代碼分析

控制中心代碼稍微複雜些，其中比較關鍵的有三個地方：

1. 使用python的uuid來產生惟一的correlation_id。
2. 發送計算請求時，設定參數correlation_id。
3. 定義一個字典來保存返回的數據，而且鍵值爲相應線程產生的correlation_id。

代碼以下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66

#!/usr/bin/env python #coding=utf8 import pika, threading, uuid   #自定義線程類，繼承threading.Thread class MyThread(threading.Thread):     def __init__(self, func, num):         super(MyThread, self).__init__()         self.func = func         self.num = num       def run(self):         print " [x] Requesting increase(%d)" % self.num         response = self.func(self.num)         print " [.] increase(%d)=%d" % (self.num, response)   #控制中心類 class Center(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)           #返回的結果都會存儲在該字典裏         self.response = {}       #定義接收到返回消息的處理方法     def on_response(self, ch, method, props, body):         self.response[props.correlation_id] = body       def request(self, n):         corr_id = str(uuid.uuid4())         self.response[corr_id] = None           #發送計算請求，並設定返回隊列和correlation_id         self.channel.basic_publish(exchange='',                                    routing_key='compute_queue',                                    properties=pika.BasicProperties(                                          reply_to = self.callback_queue,                                          correlation_id = corr_id,                                          ),                                    body=str(n))         #接收返回的數據         while self.response[corr_id] is None:             self.connection.process_data_events()         return int(self.response[corr_id])   center = Center() #發起5次計算請求 nums= [10, 20, 30, 40 ,50] threads = [] for num in nums:     threads.append(MyThread(center.request, num)) for thread in threads:     thread.start() for thread in threads:     thread.join()

筆者開啓了兩個終端，來運行compute.py，開啓一個終端來運行center.py，最後結果輸出截圖以下：

python使用rabbitmq多節點結果返回圖示

能夠看到雖然獲取的結果不是順序輸出，可是結果和源數據都是對應的。

這邊示例的作法就是建立一個隊列，使用correlation id來標識每次請求。也有作法能夠不使用correlation id，就是每請求一次，就建立一個臨時隊列，不過這樣太消耗性能了，官方也不推薦這麼作。