詳解python分布式進(jìn)程
在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ù)：
# 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ò)訪問(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)也可以）：
# 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ò)訪問(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()
任務(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é)果?，F(xiàn)在啟動(dòng)task_worker.py進(jìn)程：
$ python3 task_worker.pyConnect 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è)簡(jiǎn)單的Master/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之所以能通過(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ī)器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連接不上。
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)上讀取文件。
以上就是本篇文章所講述的所有內(nèi)容，這篇文章主要介紹了python分布式進(jìn)程的相關(guān)知識(shí)，希望你能借助資料從而理解上述所說(shuō)的內(nèi)容。希望我在這片文章所講述的內(nèi)容能夠?qū)δ阌兴鶐椭?，讓你學(xué)習(xí)python更加輕松。
相關(guān)文章
Python列表轉(zhuǎn)一維DataFrame的完整指南
在數(shù)據(jù)處理領(lǐng)域,Pandas的DataFrame是當(dāng)之無(wú)愧的王者,本文將用5個(gè)核心方法,教你優(yōu)雅地將一維列表轉(zhuǎn)換為Pandas DataFrame,感興趣的可以了解下2025-04-04
余弦相似性計(jì)算及python代碼實(shí)現(xiàn)過(guò)程解析
這篇文章主要介紹了余弦相似性計(jì)算及python代碼實(shí)現(xiàn)過(guò)程解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-09-09
pyecharts調(diào)整圖例與各板塊的位置間距實(shí)例
這篇文章主要介紹了pyecharts調(diào)整圖例與各板塊的位置間距實(shí)例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-05-05
在本篇文章里小編給大家整理的是一篇關(guān)于python中最小二乘法詳細(xì)講解內(nèi)容，需要的朋友們可以參考下。2021-02-02
Python?Pygame實(shí)戰(zhàn)之紅心大戰(zhàn)游戲的實(shí)現(xiàn)
說(shuō)起Windows自帶的游戲，相信許多80、90后的朋友都不陌生。本文就將利用Python中的Pygame模塊實(shí)現(xiàn)一下windows經(jīng)典游戲之一的紅心大戰(zhàn)，需要的可以參考一下2022-02-02
開(kāi)啟Django博客的RSS功能的實(shí)現(xiàn)方法
這篇文章主要介紹了開(kāi)啟Django博客的RSS功能的實(shí)現(xiàn)方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-02-02
這篇文章主要介紹了Python中的Classes和Metaclasses詳解,屬于基礎(chǔ)知識(shí)中類(lèi)與對(duì)象的概念部分的深入,需要的朋友可以參考下2015-04-04
Python tkinter實(shí)現(xiàn)日期選擇器
這篇文章主要為大家詳細(xì)介紹了Python tkinter實(shí)現(xiàn)日期選擇器，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-02-02
python多項(xiàng)式擬合之np.polyfit 和 np.polyld詳解
這篇文章主要介紹了python多項(xiàng)式擬合之np.polyfit 和 np.polyld的實(shí)例代碼，python數(shù)據(jù)擬合主要可采用numpy庫(kù)，庫(kù)的安裝可直接用pip install numpy等,需要的朋友跟隨小編一起學(xué)習(xí)吧2020-02-02
在項(xiàng)目中前后端數(shù)據(jù)相互是一種常態(tài),下面這篇文章主要給大家介紹了關(guān)于Django中使用AJAX的詳細(xì)過(guò)程,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2022-11-11