在Python程序中實現(xiàn)分布式進程的教程
在Thread和Process中，應(yīng)當優(yōu)選Process，因為Process更穩(wěn)定，而且，Process可以分布到多臺機器上，而Thread最多只能分布到同一臺機器的多個CPU上。
Python的multiprocessing模塊不但支持多進程，其中managers子模塊還支持把多進程分布到多臺機器上。一個服務(wù)進程可以作為調(diào)度者，將任務(wù)分布到其他多個進程中，依靠網(wǎng)絡(luò)通信。由于managers模塊封裝很好，不必了解網(wǎng)絡(luò)通信的細節(jié)，就可以很容易地編寫分布式多進程程序。
舉個例子：如果我們已經(jīng)有一個通過Queue通信的多進程程序在同一臺機器上運行，現(xiàn)在，由于處理任務(wù)的進程任務(wù)繁重，希望把發(fā)送任務(wù)的進程和處理任務(wù)的進程分布到兩臺機器上。怎么用分布式進程實現(xiàn)？
原有的Queue可以繼續(xù)使用，但是，通過managers模塊把Queue通過網(wǎng)絡(luò)暴露出去，就可以讓其他機器的進程訪問Queue了。
我們先看服務(wù)進程，服務(wù)進程負責啟動Queue，把Queue注冊到網(wǎng)絡(luò)上，然后往Queue里面寫入任務(wù)：
# taskmanager.py
import random, time, Queue
from multiprocessing.managers import BaseManager
# 發(fā)送任務(wù)的隊列:
task_queue = Queue.Queue()
# 接收結(jié)果的隊列:
result_queue = Queue.Queue()
# 從BaseManager繼承的QueueManager:
class QueueManager(BaseManager):
pass
# 把兩個Queue都注冊到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 綁定端口5000, 設(shè)置驗證碼'abc':
manager = QueueManager(address=('', 5000), authkey='abc')
# 啟動Queue:
manager.start()
# 獲得通過網(wǎng)絡(luò)訪問的Queue對象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放幾個任務(wù)進去:
for i in range(10):
n = random.randint(0, 10000)
print('Put task %d...' % n)
task.put(n)
# 從result隊列讀取結(jié)果:
print('Try get results...')
for i in range(10):
r = result.get(timeout=10)
print('Result: %s' % r)
# 關(guān)閉:
manager.shutdown()
請注意，當我們在一臺機器上寫多進程程序時，創(chuàng)建的Queue可以直接拿來用，但是，在分布式多進程環(huán)境下，添加任務(wù)到Queue不可以直接對原始的task_queue進行操作，那樣就繞過了QueueManager的封裝，必須通過manager.get_task_queue()獲得的Queue接口添加。
然后，在另一臺機器上啟動任務(wù)進程（本機上啟動也可以）：
# taskworker.py
import time, sys, Queue
from multiprocessing.managers import BaseManager
# 創(chuàng)建類似的QueueManager:
class QueueManager(BaseManager):
pass
# 由于這個QueueManager只從網(wǎng)絡(luò)上獲取Queue，所以注冊時只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')
# 連接到服務(wù)器，也就是運行taskmanager.py的機器:
server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
# 端口和驗證碼注意保持與taskmanager.py設(shè)置的完全一致:
m = QueueManager(address=(server_addr, 5000), authkey='abc')
# 從網(wǎng)絡(luò)連接:
m.connect()
# 獲取Queue的對象:
task = m.get_task_queue()
result = m.get_result_queue()
# 從task隊列取任務(wù),并把結(jié)果寫入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.')
# 處理結(jié)束:
print('worker exit.')
任務(wù)進程要通過網(wǎng)絡(luò)連接到服務(wù)進程，所以要指定服務(wù)進程的IP。
現(xiàn)在，可以試試分布式進程的工作效果了。先啟動taskmanager.py服務(wù)進程：
$ 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進程發(fā)送完任務(wù)后，開始等待result隊列的結(jié)果?，F(xiàn)在啟動taskworker.py進程：
$ python taskworker.py 127.0.0.1 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.
taskworker進程結(jié)束，在taskmanager進程中會繼續(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
這個簡單的Manager/Worker模型有什么用？其實這就是一個簡單但真正的分布式計算，把代碼稍加改造，啟動多個worker，就可以把任務(wù)分布到幾臺甚至幾十臺機器上，比如把計算n*n的代碼換成發(fā)送郵件，就實現(xiàn)了郵件隊列的異步發(fā)送。
Queue對象存儲在哪？注意到taskworker.py中根本沒有創(chuàng)建Queue的代碼，所以，Queue對象存儲在taskmanager.py進程中：
而Queue之所以能通過網(wǎng)絡(luò)訪問，就是通過QueueManager實現(xiàn)的。由于QueueManager管理的不止一個Queue，所以，要給每個Queue的網(wǎng)絡(luò)調(diào)用接口起個名字，比如get_task_queue。
authkey有什么用？這是為了保證兩臺機器正常通信，不被其他機器惡意干擾。如果taskworker.py的authkey和taskmanager.py的authkey不一致，肯定連接不上。
小結(jié)
Python的分布式進程接口簡單，封裝良好，適合需要把繁重任務(wù)分布到多臺機器的環(huán)境下。
注意Queue的作用是用來傳遞任務(wù)和接收結(jié)果，每個任務(wù)的描述數(shù)據(jù)量要盡量小。比如發(fā)送一個處理日志文件的任務(wù)，就不要發(fā)送幾百兆的日志文件本身，而是發(fā)送日志文件存放的完整路徑，由Worker進程再去共享的磁盤上讀取文件。
相關(guān)文章
Python+騰訊云服務(wù)器實現(xiàn)每日自動健康打卡
本文主要介紹了通過Python+騰訊云服務(wù)器實現(xiàn)每日自動健康打卡，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習或者工作具有一定的參考學(xué)習價值，需要的朋友們下面隨著小編來一起學(xué)習學(xué)習吧2021-12-12
在面向?qū)ο蟮睦碚撝?，有兩個核心的概念：類和實例。在?Python?里面，類和實例都是對象，也就是所謂的類對象和實例對象。本文主要來大家重溫一下Python對象模型的相關(guān)知識，感興趣的可以了解一下2023-04-04
Python讀取大型數(shù)據(jù)文件的6種方式匯總
在 Python 中，我們可以使用多種方法讀取大型數(shù)據(jù)文件，本文主要為大家介紹6個常用的Python讀取大型數(shù)據(jù)文件的方法，希望對大家有所幫助2023-05-05
今天小編就為大家分享一篇python創(chuàng)建n行m列數(shù)組示例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-12-12
html庫是用于解析HTML的一個工具，是python自帶的標準庫之一，今天通過本文給大家介紹下python中的標準庫html，感興趣的朋友一起看看吧2022-04-04
Python排序方法中sort和sorted的區(qū)別詳解
在python中常用的排序函數(shù)就是sort()和sorted()這兩個函數(shù),使用 sort() 或內(nèi)建函數(shù) sorted() 對列表進行排序,本文將詳細介紹sorted和sort兩者之間的區(qū)別,感興趣的可以了解一下2023-08-08
Python讀取mat(matlab數(shù)據(jù)文件)并實現(xiàn)畫圖
這篇文章主要介紹了Python讀取mat(matlab數(shù)據(jù)文件)并實現(xiàn)畫圖問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-12-12
Python?selenium?get_cookies獲取cookie不全的解決方案
之前使用爬蟲時最讓我頭疼的就是cookie失效的問題了,下面這篇文章主要給大家介紹了關(guān)于Python?selenium?get_cookies獲取cookie不全的解決方案,需要的朋友可以參考下2022-10-10
這篇文章主要為大家詳細介紹了Python筆記之代理模式，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-11-11
python使用jenkins發(fā)送企業(yè)微信通知的實現(xiàn)
公司使用的是企業(yè)微信，因此考慮Jenkins通知企業(yè)微信機器人的實現(xiàn)方式，本文主要介紹了python使用jenkins發(fā)送企業(yè)微信通知的實現(xiàn)，感興趣的可以了解一下2021-06-06