Python分佈式進程中你會遇到的坑

DBnXd.png

寫在前面

小驚大怪

你是否是在用Python3或者在windows系統上編程?最重要的是你對進程和線程不是很清楚?那麼恭喜你,在python分佈式進程中,會有坑等着你去挖。。。(hahahaha,此處容許我嚇唬一下你)開玩笑的啦,不過,若是你知道序列中不支持匿名函數,那這個坑就和你say byebye了。好了話很少數,直接進入正題。python

分佈式進程

正如你們所知道的Process比Thread更穩定,並且Process能夠分佈到多臺機器上,而Thread最多隻能分佈到同一臺機器的多個CPU上。Python的multiprocessing模塊不但支持多進程,其中managers子模塊還支持把多進程分佈到多臺機器上。一個服務進程能夠做爲調度者,將任務分佈到其餘多個進程中,依靠網絡通訊。因爲managers模塊封裝很好,沒必要了解網絡通訊的細節,就能夠很容易地編寫分佈式多進程程序。linux

代碼記錄

舉個例子

若是咱們已經有一個經過Queue通訊的多進程程序在同一臺機器上運行,如今,因爲處理任務的進程任務繁重,但願把發送任務的進程和處理任務的進程分佈到兩臺機器上,這應該怎麼用分佈式進程來實現呢?你已經知道了原有的Queue能夠繼續使用,並且經過managers模塊把Queue經過網絡暴露出去,就可讓其餘機器的進程來訪問Queue了。好,那咱們就這麼幹!編程

寫個task_master.py

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

#!/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

# 發送任務的隊列:
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

#!/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

# 建立相似的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服務進程:hexo

Traceback (most recent call last):
  File "F:/Python/untitled/xianchengjincheng/master.py", line 25, in <module>
    manager.start()
  File "F:\Python\pystall\lib\multiprocessing\managers.py", line 513, in start
    self._process.start()
  File "F:\Python\pystall\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "F:\Python\pystall\lib\multiprocessing\context.py", line 322, in _Popen
    return Popen(process_obj)
  File "F:\Python\pystall\lib\multiprocessing\popen_spawn_win32.py", line 65, in __init__
    reduction.dump(process_obj, to_child)
  File "F:\Python\pystall\lib\multiprocessing\reduction.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進程發送完任務後,開始等待result隊列的結果。如今啓動task_worker.py進程:dom

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:\Python\pystall\lib\multiprocessing\managers.py", line 489, in connect
    conn = Client(self._address, authkey=self._authkey)
  File "F:\Python\pystall\lib\multiprocessing\connection.py", line 487, in Client
    c = SocketClient(address)
  File "F:\Python\pystall\lib\multiprocessing\connection.py", line 614, in SocketClient
    s.connect(address)
ConnectionRefusedError: [WinError 10061] 因爲目標計算機積極拒絕,沒法鏈接。

看到沒,結果都出錯了,咱們好好分析一下到底哪出錯了。。。異步

錯誤分析

在task_master.py的報錯提示中,咱們知道它說lambda錯誤,這是由於序列化不支持匿名函數,因此咱們得修改代碼,從新對queue用QueueManager進行封裝放到網絡中。分佈式

# 把兩個Queue都註冊到網絡上, callable參數關聯了Queue對象
QueueManager.register('get_task_queue',callable=return_task_queue)   
QueueManager.register('get_result_queue',callable=return_result_queue)

其中task_queue和result_queue是兩個隊列,分別存聽任務和結果。它們用來進行進程間通訊,交換對象。 由於是分佈式的環境,放入queue中的數據須要等待Workers機器運算處理後再進行讀取,這樣就須要對queue用QueueManager進行封裝放到網絡中,這是經過上面的2行代碼來實現的。咱們給return_task_queue的網絡調用接口取了一個名get_task_queue,而return_result_queue的名字是get_result_queue,方便區分對哪一個queue進行操做。task.put(n)便是對task_queue進行寫入數據,至關於分配任務。而result.get()便是等待workers機器處理後返回的結果。

值得注意 在windows系統中你必需要寫IP地址,而其餘操做系統好比linux操做系統則就不要了。

# 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()  # 發送任務的隊列:
result_queue = queue.Queue() # 接收結果的隊列:
class QueueManager(BaseManager):  # 從BaseManager繼承的QueueManager:
    pass
# windows下運行
def return_task_queue():
    global task_queue
    return task_queue  # 返回發送任務隊列
def return_result_queue ():
    global result_queue
    return result_queue # 返回接收結果隊列

def test():
    # 把兩個Queue都註冊到網絡上, callable參數關聯了Queue對象,它們用來進行進程間通訊,交換對象
    #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, 設置驗證碼'abc':
    #manager = QueueManager(address=('', 5000), authkey=b'abc')
    # windows須要寫ip地址
    manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
    manager.start()  # 啓動Queue:
    # 得到經過網絡訪問的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):
        # 這裏加了異常捕獲
        try:
            r = result.get(timeout=5)
            print('Result: %s' % r)
        except queue.Empty:
             print('result queue is empty.')
    # 關閉:
    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

# 建立相似的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.')

先運行task_master.py,而後再運行task_worker.py (1)task_master.py運行結果以下

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運行結果以下

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.

知識補充

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

Queue對象存儲在哪?注意到task_worker.py中根本沒有建立Queue的代碼,因此,Queue對象存儲在task_master.py進程中:DBVf6.png

而Queue之因此能經過網絡訪問,就是經過QueueManager實現的。因爲QueueManager管理的不止一個Queue,因此,要給每一個Queue的網絡調用接口起個名字,好比get_task_queue。task_worker這裏的QueueManager註冊的名字必須和task_manager中的同樣。對比上面的例子,能夠看出Queue對象從另外一個進程經過網絡傳遞了過來。只不過這裏的傳遞和網絡通訊由QueueManager完成。

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

今天的分享就到這裏了,若是你有任何不懂的問題,能夠發信息或者留言嘍。

相關文章
相關標籤/搜索