[Python] 分佈式進程

轉自：分佈式進程 - 廖雪峯的官方網站

 

在Thread和Process中，應當優選Process，由於Process更穩定，並且，Process能夠分佈到多臺機器上，而Thread最多隻能分佈到同一臺機器的多個CPU上。

Python的 multiprocessing 模塊不但支持多進程，其中 managers 子模塊還支持把多進程分佈到多臺機器上。一個服務進程能夠做爲調度者，將任務分佈到其餘多個進程中，依靠網絡通訊。因爲 managers 模塊封裝很好，沒必要了解網絡通訊的細節，就能夠很容易地編寫分佈式多進程程序。

 

舉個例子：若是咱們已經有一個經過 Queue 通訊的多進程程序在同一臺機器上運行，如今，因爲處理任務的進程任務繁重，但願把發送任務的進程和處理任務的進程分佈到兩臺機器上。怎麼用分佈式進程實現？

原有的Queue能夠繼續使用，可是，經過managers模塊把Queue經過網絡暴露出去，就可讓其餘機器的進程訪問Queue了。

 

咱們先看服務進程，服務進程負責啓動Queue，把Queue註冊到網絡上，而後往Queue裏面寫入任務：

# task_master.py

import random, time, queue
from multiprocessing.managers import BaseManager

# 發送任務的隊列:
task_queue = queue.Queue()
# 接收結果的隊列:
result_queue = queue.Queue()

# 從BaseManager繼承的QueueManager:
class QueueManager(BaseManager):
    pass

# 把兩個Queue都註冊到網絡上, callable參數關聯了Queue對象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 綁定端口5000, 設置驗證碼'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 啓動Queue:
manager.start()
# 得到經過網絡訪問的Queue對象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放幾個任務進去:
for i in range(10):
    n = random.randint(0, 10000)
    print('Put task %d...' % n)
    task.put(n)
# 從result隊列讀取結果:
print('Try get results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result: %s' % r)
# 關閉:
manager.shutdown()
print('master exit.')

 請注意，當咱們在一臺機器上寫多進程程序時，建立的 Queue 能夠直接拿來用，可是，在分佈式多進程環境下，添加任務到 Queue 不能夠直接對原始的 task_queue 進行操做，那樣就繞過了 QueueManager 的封裝，必須經過 manager.get_task_queue() 得到的 Queue 接口添加。

 

而後，在另外一臺機器上啓動任務進程（本機上啓動也能夠）：

# task_worker.py

import time, sys, queue
from multiprocessing.managers import BaseManager

# 建立相似的QueueManager:
class QueueManager(BaseManager):
    pass

# 因爲這個QueueManager只從網絡上獲取Queue，因此註冊時只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

# 鏈接到服務器，也就是運行task_master.py的機器:
server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
# 端口和驗證碼注意保持與task_master.py設置的徹底一致:
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
# 從網絡鏈接:
m.connect()
# 獲取Queue的對象:
task = m.get_task_queue()
result = m.get_result_queue()
# 從task隊列取任務,並把結果寫入result隊列:
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.')

 任務進程要經過網絡鏈接到服務進程，因此要指定服務進程的IP。

 

如今，能夠試試分佈式進程的工做效果了。先啓動 task_master.py 服務進程：

$ 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進程發送完任務後，開始等待result隊列的結果。如今啓動 task_worker.py 進程：

$ 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 進程結束，在 task_master.py 進程中會繼續打印出結果：

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

 

這個簡單的Master/Worker模型有什麼用？其實這就是一個簡單但真正的分佈式計算，把代碼稍加改造，啓動多個worker，就能夠把任務分佈到幾臺甚至幾十臺機器上，好比把計算 n*n 的代碼換成發送郵件，就實現了郵件隊列的異步發送。

Queue對象存儲在哪？注意到 task_worker.py 中根本沒有建立Queue的代碼，因此，Queue對象存儲在 task_master.py 進程中：

而 Queue 之因此能經過網絡訪問，就是經過 QueueManager 實現的。因爲 QueueManager 管理的不止一個 Queue ，因此，要給每一個 Queue 的網絡調用接口起個名字，好比 get_task_queue 。

 authkey 有什麼用？這是爲了保證兩臺機器正常通訊，不被其餘機器惡意干擾。若是 task_worker.py 的 authkey 和 task_master.py 的 authkey 不一致，確定鏈接不上。

 

小結
Python的分佈式進程接口簡單，封裝良好，適合須要把繁重任務分佈到多臺機器的環境下。

注意Queue的做用是用來傳遞任務和接收結果，每一個任務的描述數據量要儘可能小。好比發送一個處理日誌文件的任務，就不要發送幾百兆的日誌文件自己，而是發送日誌文件存放的完整路徑，由Worker進程再去共享的磁盤上讀取文件。