這篇文章主要介紹了Python進(jìn)行分布式進(jìn)程的方法，具有一定借鑒價(jià)值，需要的朋友可以參考下。希望大家閱讀完這篇文章后大有收獲。下面讓小編帶著大家一起了解一下。
成都創(chuàng)新互聯(lián)是一家專注于網(wǎng)站設(shè)計(jì)制作、成都做網(wǎng)站與策劃設(shè)計(jì),沙河網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)做網(wǎng)站,專注于網(wǎng)站建設(shè)10余年,網(wǎng)設(shè)計(jì)領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:沙河等地區(qū)。沙河做網(wǎng)站價(jià)格咨詢:13518219792
分布式進(jìn)程
在Thread和Process中，應(yīng)當(dāng)優(yōu)選Process，因?yàn)镻rocess更穩(wěn)定，而且，Process可以分布到多臺機(jī)器上，而Thread最多只能分布到同一臺機(jī)器的多個(gè)CPU上。
Python的multiprocessing模塊不但支持多進(jìn)程，其中managers子模塊還支持把多進(jìn)程分布到多臺機(jī)器上。一個(gè)服務(wù)進(jìn)程可以作為調(diào)度者，將任務(wù)分布到其他多個(gè)進(jìn)程中，依靠網(wǎng)絡(luò)通信。由于managers模塊封裝很好，不必了解網(wǎng)絡(luò)通信的細(xì)節(jié)，就可以很容易地編寫分布式多進(jìn)程程序。
舉個(gè)例子：如果我們已經(jīng)有一個(gè)通過Queue通信的多進(jìn)程程序在同一臺機(jī)器上運(yùn)行，現(xiàn)在，由于處理任務(wù)的進(jìn)程任務(wù)繁重，希望把發(fā)送任務(wù)的進(jìn)程和處理任務(wù)的進(jìn)程分布到兩臺機(jī)器上。怎么用分布式進(jìn)程實(shí)現(xiàn)？
原有的Queue可以繼續(xù)使用，但是，通過managers模塊把Queue通過網(wǎng)絡(luò)暴露出去，就可以讓其他機(jī)器的進(jìn)程訪問Queue了。
我們先看服務(wù)進(jìn)程，服務(wù)進(jìn)程負(fù)責(zé)啟動(dòng)Queue，把Queue注冊到網(wǎng)絡(luò)上，然后往Queue里面寫入任務(wù)：
重慶分公司
重慶分公司
# _*_ coding:utf-8 _*_
import random, time, queue
from multiprocessing.managers import BaseManager
# the queue of send tasks
task_queue = queue.Queue()
# the queue of recive tasks
result_queue = queue.Queue()
# cong BaseManager jicheng de QueueManager
class QueueManager(BaseManager):
pass
# 把兩個(gè)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è)置驗(yàn)證碼'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 啟動(dòng)Queue
manager.start()
# 獲得通過網(wǎng)絡(luò)訪問的Queue對象
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)
# Close
manager.shutdown()
print('master exit.')請注意，當(dāng)我們在一臺機(jī)器上寫多進(jìn)程程序時(shí)，創(chuàng)建的Queue可以直接拿來用，但是，在分布式多進(jìn)程環(huán)境下，添加任務(wù)到Queue不可以直接對原始的task_queue進(jìn)行操作，那樣就繞過了QueueManager的封裝，必須通過manager.get_task_queue()獲得的Queue接口添加。
然后，在另一臺機(jī)器上啟動(dòng)任務(wù)進(jìn)程（本機(jī)上啟動(dòng)也可以）：
# task_worker.py
import time, sys, queue
from multiprocessing.managers import BaseManager
# 創(chuàng)建類似的QueueManager:
class QueueManager(BaseManager):
pass
# 由于這個(gè)QueueManager只從網(wǎng)絡(luò)上獲取Queue，所以注冊時(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的對象:
task = m.get_task_queue()
result = m.get_result_queue()
# 從task隊(duì)列取任務(wù),并把結(jié)果寫入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)程要通過網(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ù)后，開始等待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
這個(gè)簡單的Master/Worker模型有什么用？其實(shí)這就是一個(gè)簡單但真正的分布式計(jì)算，把代碼稍加改造，啟動(dòng)多個(gè)worker，就可以把任務(wù)分布到幾臺甚至幾十臺機(jī)器上，比如把計(jì)算n*n的代碼換成發(fā)送郵件，就實(shí)現(xiàn)了郵件隊(duì)列的異步發(fā)送。
Queue對象存儲(chǔ)在哪？注意到task_worker.py中根本沒有創(chuàng)建Queue的代碼，所以，Queue對象存儲(chǔ)在task_master.py進(jìn)程中。
而Queue之所以能通過網(wǎng)絡(luò)訪問，就是通過QueueManager實(shí)現(xiàn)的。由于QueueManager管理的不止一個(gè)Queue，所以，要給每個(gè)Queue的網(wǎng)絡(luò)調(diào)用接口起個(gè)名字，比如get_task_queue。
authkey有什么用？這是為了保證兩臺機(jī)器正常通信，不被其他機(jī)器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連接不上。
Python的分布式進(jìn)程接口簡單，封裝良好，適合需要把繁重任務(wù)分布到多臺機(jī)器的環(huán)境下。
注意Queue的作用是用來傳遞任務(wù)和接收結(jié)果，每個(gè)任務(wù)的描述數(shù)據(jù)量要盡量小。比如發(fā)送一個(gè)處理日志文件的任務(wù)，就不要發(fā)送幾百兆的日志文件本身，而是發(fā)送日志文件存放的完整路徑，由Worker進(jìn)程再去共享的磁盤上讀取文件。
感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享Python進(jìn)行分布式進(jìn)程的方法內(nèi)容對大家有幫助，同時(shí)也希望大家多多支持創(chuàng)新互聯(lián)，關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，遇到問題就找創(chuàng)新互聯(lián)，詳細(xì)的解決方法等著你來學(xué)習(xí)!
