先說一下筆者這里的測試環(huán)境：Ubuntu14.04 + Python 2.7.4
RabbitMQ服務器
|
1
|
sudo apt-get install rabbitmq-server |
Python使用RabbitMQ需要Pika庫
|
1
|
sudo pip install pika |
遠程結果返回
消息發(fā)送端發(fā)送消息出去后沒有結果返回。如果只是單純發(fā)送消息，當然沒有問題了，但是在實際中，常常會需要接收端將收到的消息進行處理之后，返回給發(fā)送端。
處理方法描述：發(fā)送端在發(fā)送信息前，產生一個接收消息的臨時隊列，該隊列用來接收返回的結果。其實在這里接收端、發(fā)送端的概念已經比較模糊了，因為發(fā)送端也同樣要接收消息，接收端同樣也要發(fā)送消息，所以這里筆者使用另外的示例來演示這一過程。
示例內容：假設有一個控制中心和一個計算節(jié)點，控制中心會將一個自然數(shù)N發(fā)送給計算節(jié)點，計算節(jié)點將N值加1后，返回給控制中心。這里用center.py模擬控制中心，compute.py模擬計算節(jié)點。
compute.py代碼分析
|
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
|
#!/usr/bin/env python#coding=utf8import pika #連接rabbitmq服務器connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))channel = connection.channel() #定義隊列channel.queue_declare(queue='compute_queue')print ' [*] Waiting for n' #將n值加1def increase(n): return n + 1 #定義接收到消息的處理方法def request(ch, method, properties, body): print " [.] increase(%s)" % (body,) response = increase(int(body)) #將計算結果發(fā)送回控制中心 ch.basic_publish(exchange='', routing_key=properties.reply_to, 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() |
計算節(jié)點的代碼比較簡單，值得一提的是，原來的接收方法都是直接將消息打印出來，這邊進行了加一的計算，并將結果發(fā)送回控制中心。
center.py代碼分析
|
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
|
#!/usr/bin/env python#coding=utf8import pika 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) #定義接收到返回消息的處理方法 def on_response(self, ch, method, props, body): self.response = body def request(self, n): self.response = None #發(fā)送計算請求，并聲明返回隊列 self.channel.basic_publish(exchange='', routing_key='compute_queue', properties=pika.BasicProperties( reply_to = self.callback_queue, ), body=str(n)) #接收返回的數(shù)據(jù) while self.response is None: self.connection.process_data_events() return int(self.response) center = Center() print " [x] Requesting increase(30)"response = center.request(30)print " [.] Got %r" % (response,) |
上例代碼定義了接收返回數(shù)據(jù)的隊列和處理方法，并且在發(fā)送請求的時候將該隊列賦值給reply_to，在計算節(jié)點代碼中就是通過這個參數(shù)來獲取返回隊列的。
打開兩個終端，一個運行代碼python compute.py，另外一個終端運行center.py，如果執(zhí)行成功，應該就能看到效果了。
筆者在測試的時候，出了些小問題，就是在center.py發(fā)送消息時沒有指明返回隊列，結果compute.py那邊在計算完結果要發(fā)回數(shù)據(jù)時報錯，提示routing_key不存在，再次運行也報錯。用rabbitmqctl list_queues查看隊列，發(fā)現(xiàn)compute_queue隊列有1條數(shù)據(jù)，每次重新運行compute.py的時候，都會重新處理這條數(shù)據(jù)。后來使用/etc/init.d/rabbitmq-server restart重新啟動下rabbitmq就ok了。
相互關聯(lián)編號correlation id
上一遍演示了遠程結果返回的示例，但是有一個沒有提到，就是correlation id，這個是個什么東東呢？
假設有多個計算節(jié)點，控制中心開啟多個線程，往這些計算節(jié)點發(fā)送數(shù)字，要求計算結果并返回，但是控制中心只開啟了一個隊列，所有線程都是從這個隊列里獲取消息，每個線程如何確定收到的消息就是該線程對應的呢？這個就是correlation id的用處了。correlation翻譯成中文就是相互關聯(lián)，也表達了這個意思。
correlation id運行原理：控制中心發(fā)送計算請求時設置correlation id，而后計算節(jié)點將計算結果，連同接收到的correlation id一起返回，這樣控制中心就能通過correlation id來標識請求。其實correlation id也可以理解為請求的唯一標識碼。
示例內容：控制中心開啟多個線程，每個線程都發(fā)起一次計算請求，通過correlation id，每個線程都能準確收到相應的計算結果。
compute.py代碼分析
和上面一篇相比，只需修改一個地方：將計算結果發(fā)送回控制中心時，增加參數(shù)correlation_id的設定，該參數(shù)的值其實是從控制中心發(fā)送過來的，這里只是再次發(fā)送回去。代碼如下：
|
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=utf8import pika #連接rabbitmq服務器connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))channel = connection.channel() #定義隊列channel.queue_declare(queue='compute_queue')print ' [*] Waiting for n' #將n值加1def increase(n): return n + 1 #定義接收到消息的處理方法def request(ch, method, props, body): print " [.] increase(%s)" % (body,) response = increase(int(body)) #將計算結果發(fā)送回控制中心，增加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代碼分析
控制中心代碼稍微復雜些，其中比較關鍵的有三個地方：
使用python的uuid來產生唯一的correlation_id。
發(fā)送計算請求時，設定參數(shù)correlation_id。
定義一個字典來保存返回的數(shù)據(jù)，并且鍵值為相應線程產生的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=utf8import pika, threading, uuid #自定義線程類，繼承threading.Threadclass 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 #發(fā)送計算請求，并設定返回隊列和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)) #接收返回的數(shù)據(jù) while self.response[corr_id] is None: self.connection.process_data_events() return int(self.response[corr_id]) center = Center()#發(fā)起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，最后結果輸出截圖如下：
可以看到雖然獲取的結果不是順序輸出，但是結果和源數(shù)據(jù)都是對應的。
這邊示例的做法就是創(chuàng)建一個隊列，使用correlation id來標識每次請求。也有做法可以不使用correlation id，就是每請求一次，就創(chuàng)建一個臨時隊列，不過這樣太消耗性能了，官方也不推薦這么做。
