在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)程訪(fǎng)問(wèn)Queue了柜蜈。
我們先看服務(wù)進(jìn)程，服務(wù)進(jìn)程負(fù)責(zé)啟動(dòng)Queue指巡，把Queue注冊(cè)到網(wǎng)絡(luò)上淑履，然后往Queue里面寫(xiě)入任務(wù)：
# task_master.py
import random, time, queue from 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=b'abc')
# 啟動(dòng)Queue:
manager.start()
# 獲得通過(guò)網(wǎng)絡(luò)訪(fǎng)問(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() print('master exit.')
請(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)也可以）：
# task_worker.py
import time, sys, queue from 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)行task_master.py的機(jī)器:
server_addr = '127.0.0.1' print('Connect to server %s...' % server_addr)
# 端口和驗(yàn)證碼注意保持與task_master.py設(shè)置的完全一致:
m = QueueManager(address=(server_addr, 5000), authkey=b'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.')
print('worker exit.')```
任務(wù)進(jìn)程要通過(guò)網(wǎng)絡(luò)連接到服務(wù)進(jìn)程崇渗，所以要指定服務(wù)進(jìn)程的IP。
現(xiàn)在，可以試試分布式進(jìn)程的工作效果了宅广。先啟動(dòng)task_master.py服務(wù)進(jìn)程：
`
$ python3 task_master.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...`
task_master.py進(jìn)程發(fā)送完任務(wù)后葫掉，開(kāi)始等待result隊(duì)列的結(jié)果。現(xiàn)在啟動(dòng)task_worker.py進(jìn)程：
`
$ python3 task_worker.py
Connect 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.
`
task_worker.py進(jìn)程結(jié)束跟狱，在task_master.py進(jìn)程中會(huì)繼續(xù)打印出結(jié)果：
`
Result: 3411 * 3411 = 11634921
Result: 1605 * 1605 = 2576025
Result: 1398 * 1398 = 1954404
Result: 4729 * 4729 = 22363441
Result: 5300 * 5300 = 28090000
Result: 7471 * 7471 = 55815841
Result: 68 * 68 = 4624
Result: 4219 * 4219 = 17799961
Result: 339 * 339 = 114921
Result: 7866 * 7866 = 61873956
`
Queue對(duì)象存儲(chǔ)在哪俭厚？注意到task_worker.py中根本沒(méi)有創(chuàng)建Queue的代碼，所以驶臊，Queue對(duì)象存儲(chǔ)在task_master.py進(jìn)程中：
而Queue之所以能通過(guò)網(wǎng)絡(luò)訪(fǎng)問(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ī)器惡意干擾爽茴。如果task_worker.py的authkey和task_master.py的authkey不一致葬凳，肯定連接不上。
