Python分布式進(jìn)程中你會遇到的問題解析
小驚大怪
你是不是在用Python3或者在windows系統(tǒng)上編程?最重要的是你對進(jìn)程和線程不是很清楚?那么恭喜你，在python分布式進(jìn)程中，會有坑等著你去挖。。。(hahahaha，此處允許我嚇唬一下你)開玩笑的啦，不過，如果你知道序列中不支持匿名函數(shù)，那這個坑就和你say byebye了。好了話不多數(shù)，直接進(jìn)入正題。
分布式進(jìn)程
正如大家所知道的Process比Thread更穩(wěn)定，而且Process可以分布到多臺機(jī)器上，而Thread最多只能分布到同一臺機(jī)器的多個CPU上。Python的multiprocessing模塊不但支持多進(jìn)程，其中managers子模塊還支持把多進(jìn)程分布到多臺機(jī)器上。一個服務(wù)進(jìn)程可以作為調(diào)度者，將任務(wù)分布到其他多個進(jìn)程中，依靠網(wǎng)絡(luò)通信。由于managers模塊封裝很好，不必了解網(wǎng)絡(luò)通信的細(xì)節(jié)，就可以很容易地編寫分布式多進(jìn)程程序。
代碼記錄
舉個例子
如果我們已經(jīng)有一個通過Queue通信的多進(jìn)程程序在同一臺機(jī)器上運(yùn)行，現(xiàn)在，由于處理任務(wù)的進(jìn)程任務(wù)繁重，希望把發(fā)送任務(wù)的進(jìn)程和處理任務(wù)的進(jìn)程分布到兩臺機(jī)器上，這應(yīng)該怎么用分布式進(jìn)程來實(shí)現(xiàn)呢?你已經(jīng)知道了原有的Queue可以繼續(xù)使用，而且通過managers模塊把Queue通過網(wǎng)絡(luò)暴露出去，就可以讓其他機(jī)器的進(jìn)程來訪問Queue了。好，那我們就這么干!
寫個task_master.py
我們先看服務(wù)進(jìn)程。服務(wù)進(jìn)程負(fù)責(zé)啟動Queue，把Queue注冊到網(wǎng)絡(luò)上，然后往Queue里面寫入任務(wù)。
#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: 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
# 把兩個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')
# 啟動Queue:
manager.start()
# 獲得通過網(wǎng)絡(luò)訪問的Queue對象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放幾個任務(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.')
請注意，當(dāng)我們在一臺機(jī)器上寫多進(jìn)程程序時，創(chuàng)建的Queue可以直接拿來用，但是，在分布式多進(jìn)程環(huán)境下，添加任務(wù)到Queue不可以直接對原始的task_queue進(jìn)行操作，那樣就繞過了QueueManager的封裝，必須通過manager.get_task_queue()獲得的Queue接口添加。然后，在另一臺機(jī)器上啟動任務(wù)進(jìn)程(本機(jī)上啟動也可以)
寫個task_worker.py
#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_worker.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ù)器，也就是運(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。
運(yùn)行結(jié)果
現(xiàn)在，可以試試分布式進(jìn)程的工作效果了。先啟動task_master.py服務(wù)進(jìn)程：
Traceback (most recent call last):
File "F:/Python/untitled/xianchengjincheng/master.py", line 25, in <module>
manager.start()
File "F:Pythonpystalllibmultiprocessingmanagers.py", line 513, in start
self._process.start()
File "F:Pythonpystalllibmultiprocessingprocess.py", line 105, in start
self._popen = self._Popen(self)
File "F:Pythonpystalllibmultiprocessingcontext.py", line 322, in _Popen
return Popen(process_obj)
File "F:Pythonpystalllibmultiprocessingpopen_spawn_win32.py", line 65, in __init__
reduction.dump(process_obj, to_child)
File "F:Pythonpystalllibmultiprocessing
eduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <function <lambda> at 0x00000202D1921E18>: attribute lookup <lambda> on __main__ failed
task_master.py進(jìn)程發(fā)送完任務(wù)后，開始等待result隊(duì)列的結(jié)果?，F(xiàn)在啟動task_worker.py進(jìn)程：
Connect to server 127.0.0.1...
Traceback (most recent call last):
File "F:/Python/untitled/xianchengjincheng/work.py", line 24, in <module>
m.connect()
File "F:Pythonpystalllibmultiprocessingmanagers.py", line 489, in connect
conn = Client(self._address, authkey=self._authkey)
File "F:Pythonpystalllibmultiprocessingconnection.py", line 487, in Client
c = SocketClient(address)
File "F:Pythonpystalllibmultiprocessingconnection.py", line 614, in SocketClient
s.connect(address)
ConnectionRefusedError: [WinError 10061] 由于目標(biāo)計(jì)算機(jī)積極拒絕，無法連接。
看到?jīng)]，結(jié)果都出錯了，我們好好分析一下到底哪出錯了。。。
錯誤分析
在task_master.py的報錯提示中，我們知道它說lambda錯誤，這是因?yàn)樾蛄谢恢С帜涿瘮?shù)，所以我們得修改代碼，重新對queue用QueueManager進(jìn)行封裝放到網(wǎng)絡(luò)中。
# 把兩個Queue都注冊到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對象
QueueManager.register('get_task_queue',callable=return_task_queue)
QueueManager.register('get_result_queue',callable=return_result_queue)
其中task_queue和result_queue是兩個隊(duì)列，分別存放任務(wù)和結(jié)果。它們用來進(jìn)行進(jìn)程間通信，交換對象。
因?yàn)槭欠植际降沫h(huán)境，放入queue中的數(shù)據(jù)需要等待Workers機(jī)器運(yùn)算處理后再進(jìn)行讀取，這樣就需要對queue用QueueManager進(jìn)行封裝放到網(wǎng)絡(luò)中，這是通過上面的2行代碼來實(shí)現(xiàn)的。我們給return_task_queue的網(wǎng)絡(luò)調(diào)用接口取了一個名get_task_queue,而return_result_queue的名字是get_result_queue，方便區(qū)分對哪個queue進(jìn)行操作。task.put(n)即是對task_queue進(jìn)行寫入數(shù)據(jù)，相當(dāng)于分配任務(wù)。而result.get()即是等待workers機(jī)器處理后返回的結(jié)果。
值得注意 在windows系統(tǒng)中你必須要寫IP地址，而其他操作系統(tǒng)比如linux操作系統(tǒng)則就不要了。
# windows需要寫ip地址
manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
修改后的代碼
在task_master.py中修改如下：
#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_master.py
# task_master.py
import random,time,queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support
task_queue = queue.Queue() # 發(fā)送任務(wù)的隊(duì)列:
result_queue = queue.Queue() # 接收結(jié)果的隊(duì)列:
class QueueManager(BaseManager): # 從BaseManager繼承的QueueManager:
pass
# windows下運(yùn)行
def return_task_queue():
global task_queue
return task_queue # 返回發(fā)送任務(wù)隊(duì)列
def return_result_queue ():
global result_queue
return result_queue # 返回接收結(jié)果隊(duì)列
def test():
# 把兩個Queue都注冊到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對象,它們用來進(jìn)行進(jìn)程間通信，交換對象
#QueueManager.register('get_task_queue', callable=lambda: task_queue)
#QueueManager.register('get_result_queue', callable=lambda: result_queue)
QueueManager.register('get_task_queue', callable=return_task_queue)
QueueManager.register('get_result_queue', callable=return_result_queue)
# 綁定端口5000, 設(shè)置驗(yàn)證碼'abc':
#manager = QueueManager(address=('', 5000), authkey=b'abc')
# windows需要寫ip地址
manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
manager.start() # 啟動Queue:
# 獲得通過網(wǎng)絡(luò)訪問的Queue對象:
task = manager.get_task_queue()
result = manager.get_result_queue()
for i in range(10): # 放幾個任務(wù)進(jìn)去:
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):
# 這里加了異常捕獲
try:
r = result.get(timeout=5)
print('Result: %s' % r)
except queue.Empty:
print('result queue is empty.')
# 關(guān)閉:
manager.shutdown()
print('master exit.')
if __name__=='__main__':
freeze_support()
print('start!')
test()
在task_worker.py中修改如下：
#！/user/bin/pytthon
# -*- coding:utf-8 -*-
# @Time: 2018/3/3 16:46
# @Author: lichexo
# @File: task_worker.py
# task_worker.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ù)器，也就是運(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.')
先運(yùn)行task_master.py,然后再運(yùn)行task_worker.py
(1)task_master.py運(yùn)行結(jié)果如下
start! Put task 7872... Put task 6931... Put task 1395... Put task 8477... Put task 8300... Put task 1597... Put task 8738... Put task 8627... Put task 1884... Put task 2561... Try get results... Result: 7872 * 7872 = 61968384 Result: 6931 * 6931 = 48038761 Result: 1395 * 1395 = 1946025 Result: 8477 * 8477 = 71859529 Result: 8300 * 8300 = 68890000 Result: 1597 * 1597 = 2550409 Result: 8738 * 8738 = 76352644 Result: 8627 * 8627 = 74425129 Result: 1884 * 1884 = 3549456 Result: 2561 * 2561 = 6558721 master exit.
(2)task_worker.py運(yùn)行結(jié)果如下
Connect to server 127.0.0.1... run task 8640 * 8640... run task 7418 * 7418... run task 9303 * 9303... run task 568 * 568... run task 1633 * 1633... run task 3583 * 3583... run task 3293 * 3293... run task 8975 * 8975... run task 8189 * 8189... run task 731 * 731... worker exit.
知識補(bǔ)充
這個簡單的Master/Worker模型有什么用?其實(shí)這就是一個簡單但真正的分布式計(jì)算，把代碼稍加改造，啟動多個worker，就可以把任務(wù)分布到幾臺甚至幾十臺機(jī)器上，比如把計(jì)算n*n的代碼換成發(fā)送郵件，就實(shí)現(xiàn)了郵件隊(duì)列的異步發(fā)送。
Queue對象存儲在哪?注意到task_worker.py中根本沒有創(chuàng)建Queue的代碼，所以，Queue對象存儲在task_master.py進(jìn)程中：
而Queue之所以能通過網(wǎng)絡(luò)訪問，就是通過QueueManager實(shí)現(xiàn)的。由于QueueManager管理的不止一個Queue，所以，要給每個Queue的網(wǎng)絡(luò)調(diào)用接口起個名字，比如get_task_queue。task_worker這里的QueueManager注冊的名字必須和task_manager中的一樣。對比上面的例子，可以看出Queue對象從另一個進(jìn)程通過網(wǎng)絡(luò)傳遞了過來。只不過這里的傳遞和網(wǎng)絡(luò)通信由QueueManager完成。
authkey有什么用?這是為了保證兩臺機(jī)器正常通信，不被其他機(jī)器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連接不上。
總結(jié)
以上所述是小編給大家介紹的Python分布式進(jìn)程中你會遇到的問題解析,希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時回復(fù)大家的。在此也非常感謝大家對腳本之家網(wǎng)站的支持！
如果你覺得本文對你有幫助，歡迎轉(zhuǎn)載，煩請注明出處，謝謝！
相關(guān)文章
如何在Django中添加沒有微秒的 DateTimeField 屬性詳解
這篇文章主要給大家介紹了關(guān)于如何在Django中添加沒有微秒的 DateTimeField 屬性的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-01-01
這篇文章主要介紹了Python中變量交換的例子,Python中的變量交換不需要臨時變量,一個等號即可實(shí)現(xiàn)變量的交換,非常方便,需要的朋友可以參考下2014-08-08
這篇文章主要用代碼幫你了解Python基礎(chǔ)，使用循環(huán)，字典和集合的示例代碼，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-01-01
python實(shí)現(xiàn)定時任務(wù)的八種方式總結(jié)
在日常工作中,我們常常會用到需要周期性執(zhí)行的任務(wù),下面這篇文章主要給大家介紹了關(guān)于python實(shí)現(xiàn)定時任務(wù)的八種方式,文中通過實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-01-01
Python內(nèi)存池機(jī)制的實(shí)現(xiàn)
Python內(nèi)存池是Python解釋器為了提高內(nèi)存分配效率而設(shè)計(jì)的一種內(nèi)存管理機(jī)制,本文主要介紹了Python內(nèi)存池機(jī)制的實(shí)現(xiàn),具有一定的參考價值,感興趣的可以了解一下2025-04-04
今天小編就為大家分享一篇Django自帶的加密算法及加密模塊詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-12-12
Python?xmltodict實(shí)現(xiàn)簡化XML數(shù)據(jù)處理
Python社區(qū)為提供了xmltodict庫,它專為簡化XML與Python數(shù)據(jù)結(jié)構(gòu)的轉(zhuǎn)換而設(shè)計(jì),本文主要來為大家介紹一下如何使用xmltodict實(shí)現(xiàn)簡化XML數(shù)據(jù)處理,希望對大家有所幫助2025-01-01
python+selenium+chromedriver實(shí)現(xiàn)爬蟲示例代碼
這篇文章主要介紹了python+selenium+chromedriver實(shí)現(xiàn)爬蟲示例代碼，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-04-04
Python實(shí)現(xiàn)多線程HTTP下載器示例
本篇文章主要介紹了Python實(shí)現(xiàn)多線程HTTP下載器示例，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-02-02
數(shù)據(jù)清洗是關(guān)鍵步驟,尤其是處理空數(shù)據(jù),Pandas提供.isna()識別、.dropna()刪除和填充函數(shù)（如使用平均值、中位數(shù)或眾數(shù)）來處理數(shù)據(jù)集中的空值,以提高數(shù)據(jù)分析的準(zhǔn)確性,有效的空數(shù)據(jù)處理可以確保數(shù)據(jù)集的完整性和可靠性2024-09-09