在Thread和Process中，應(yīng)當(dāng)優(yōu)選Process，因?yàn)镻rocess更穩(wěn)定，而且，Process可以分布到多臺(tái)機(jī)器上，而Thread最多只能分布到同一臺(tái)機(jī)器的多個(gè)CPU上。
Python的multiprocessing模塊不但支持多進(jìn)程，其中managers子模塊還支持把多進(jìn)程分布到多臺(tái)機(jī)器上。一個(gè)服務(wù)進(jìn)程可以作為調(diào)度者，將任務(wù)分布到其他多個(gè)進(jìn)程中，依靠網(wǎng)絡(luò)通信。由于managers模塊封裝很好，不必了解網(wǎng)絡(luò)通信的細(xì)節(jié)，就可以很容易地編寫(xiě)分布式多進(jìn)程程序。
舉個(gè)例子：如果我們已經(jīng)有一個(gè)通過(guò)Queue通信的多進(jìn)程程序在同一臺(tái)機(jī)器上運(yùn)行，現(xiàn)在，由于處理任務(wù)的進(jìn)程任務(wù)繁重，希望把發(fā)送任務(wù)的進(jìn)程和處理任務(wù)的進(jìn)程分布到兩臺(tái)機(jī)器上。怎么用分布式進(jìn)程實(shí)現(xiàn)？
原有的Queue可以繼續(xù)使用，但是，通過(guò)managers模塊把Queue通過(guò)網(wǎng)絡(luò)暴露出去，就可以讓其他機(jī)器的進(jìn)程訪問(wèn)Queue了。
我們先看服務(wù)進(jìn)程，服務(wù)進(jìn)程負(fù)責(zé)啟動(dòng)Queue，把Queue注冊(cè)到網(wǎng)絡(luò)上，然后往Queue里面寫(xiě)入任務(wù)：
|
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
|
# taskmanager.pyimport random, time, Queuefrom multiprocessing.managers import BaseManager# 發(fā)送任務(wù)的隊(duì)列:task_queue = Queue.Queue()# 接收結(jié)果的隊(duì)列:result_queue = Queue.Queue()# 從BaseManager繼承的QueueManager:class QueueManager(BaseManager): pass# 把兩個(gè)Queue都注冊(cè)到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對(duì)象:QueueManager.register('get_task_queue', callable=lambda: task_queue)QueueManager.register('get_result_queue', callable=lambda: result_queue)# 綁定端口5000, 設(shè)置驗(yàn)證碼'abc':manager = QueueManager(address=('', 5000), authkey='abc')# 啟動(dòng)Queue:manager.start()# 獲得通過(guò)網(wǎng)絡(luò)訪問(wèn)的Queue對(duì)象:task = manager.get_task_queue()result = manager.get_result_queue()# 放幾個(gè)任務(wù)進(jìn)去:for i in range(10): n = random.randint(0, 10000) print('Put task %d...' % n) task.put(n)# 從result隊(duì)列讀取結(jié)果:print('Try get results...')for i in range(10): r = result.get(timeout=10) print('Result: %s' % r)# 關(guān)閉:manager.shutdown() |
請(qǐng)注意，當(dāng)我們?cè)谝慌_(tái)機(jī)器上寫(xiě)多進(jìn)程程序時(shí)，創(chuàng)建的Queue可以直接拿來(lái)用，但是，在分布式多進(jìn)程環(huán)境下，添加任務(wù)到Queue不可以直接對(duì)原始的task_queue進(jìn)行操作，那樣就繞過(guò)了QueueManager的封裝，必須通過(guò)manager.get_task_queue()獲得的Queue接口添加。
然后，在另一臺(tái)機(jī)器上啟動(dòng)任務(wù)進(jìn)程（本機(jī)上啟動(dòng)也可以）：
|
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
|
# taskworker.pyimport time, sys, Queuefrom multiprocessing.managers import BaseManager# 創(chuàng)建類(lèi)似的QueueManager:class QueueManager(BaseManager): pass# 由于這個(gè)QueueManager只從網(wǎng)絡(luò)上獲取Queue，所以注冊(cè)時(shí)只提供名字:QueueManager.register('get_task_queue')QueueManager.register('get_result_queue')# 連接到服務(wù)器，也就是運(yùn)行taskmanager.py的機(jī)器:server_addr = '127.0.0.1'print('Connect to server %s...' % server_addr)# 端口和驗(yàn)證碼注意保持與taskmanager.py設(shè)置的完全一致:m = QueueManager(address=(server_addr, 5000), authkey='abc')# 從網(wǎng)絡(luò)連接:m.connect()# 獲取Queue的對(duì)象:task = m.get_task_queue()result = m.get_result_queue()# 從task隊(duì)列取任務(wù),并把結(jié)果寫(xiě)入result隊(duì)列:for i in range(10): try: n = task.get(timeout=1) print('run task %d * %d...' % (n, n)) r = '%d * %d = %d' % (n, n, n*n) time.sleep(1) result.put(r) except Queue.Empty: print('task queue is empty.')# 處理結(jié)束:print('worker exit.') |
任務(wù)進(jìn)程要通過(guò)網(wǎng)絡(luò)連接到服務(wù)進(jìn)程，所以要指定服務(wù)進(jìn)程的IP。
現(xiàn)在，可以試試分布式進(jìn)程的工作效果了。先啟動(dòng)taskmanager.py服務(wù)進(jìn)程：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
$ python taskmanager.py Put task 3411...Put task 1605...Put task 1398...Put task 4729...Put task 5300...Put task 7471...Put task 68...Put task 4219...Put task 339...Put task 7866...Try get results... |
taskmanager進(jìn)程發(fā)送完任務(wù)后，開(kāi)始等待result隊(duì)列的結(jié)果。現(xiàn)在啟動(dòng)taskworker.py進(jìn)程：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
$ python taskworker.py 127.0.0.1Connect to server 127.0.0.1...run task 3411 * 3411...run task 1605 * 1605...run task 1398 * 1398...run task 4729 * 4729...run task 5300 * 5300...run task 7471 * 7471...run task 68 * 68...run task 4219 * 4219...run task 339 * 339...run task 7866 * 7866...worker exit. |
taskworker進(jìn)程結(jié)束，在taskmanager進(jìn)程中會(huì)繼續(xù)打印出結(jié)果：
|
1
2
3
4
5
6
7
8
9
10
|
Result: 3411 * 3411 = 11634921Result: 1605 * 1605 = 2576025Result: 1398 * 1398 = 1954404Result: 4729 * 4729 = 22363441Result: 5300 * 5300 = 28090000Result: 7471 * 7471 = 55815841Result: 68 * 68 = 4624Result: 4219 * 4219 = 17799961Result: 339 * 339 = 114921Result: 7866 * 7866 = 61873956 |
這個(gè)簡(jiǎn)單的Manager/Worker模型有什么用？其實(shí)這就是一個(gè)簡(jiǎn)單但真正的分布式計(jì)算，把代碼稍加改造，啟動(dòng)多個(gè)worker，就可以把任務(wù)分布到幾臺(tái)甚至幾十臺(tái)機(jī)器上，比如把計(jì)算n*n的代碼換成發(fā)送郵件，就實(shí)現(xiàn)了郵件隊(duì)列的異步發(fā)送。
Queue對(duì)象存儲(chǔ)在哪？注意到taskworker.py中根本沒(méi)有創(chuàng)建Queue的代碼，所以，Queue對(duì)象存儲(chǔ)在taskmanager.py進(jìn)程中：
而Queue之所以能通過(guò)網(wǎng)絡(luò)訪問(wèn)，就是通過(guò)QueueManager實(shí)現(xiàn)的。由于QueueManager管理的不止一個(gè)Queue，所以，要給每個(gè)Queue的網(wǎng)絡(luò)調(diào)用接口起個(gè)名字，比如get_task_queue。
authkey有什么用？這是為了保證兩臺(tái)機(jī)器正常通信，不被其他機(jī)器惡意干擾。如果taskworker.py的authkey和taskmanager.py的authkey不一致，肯定連接不上。
小結(jié)
Python的分布式進(jìn)程接口簡(jiǎn)單，封裝良好，適合需要把繁重任務(wù)分布到多臺(tái)機(jī)器的環(huán)境下。
注意Queue的作用是用來(lái)傳遞任務(wù)和接收結(jié)果，每個(gè)任務(wù)的描述數(shù)據(jù)量要盡量小。比如發(fā)送一個(gè)處理日志文件的任務(wù)，就不要發(fā)送幾百兆的日志文件本身，而是發(fā)送日志文件存放的完整路徑，由Worker進(jìn)程再去共享的磁盤(pán)上讀取文件。
