[[266539]]

小驚大怪

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

分布式進程

正如大家所知道的Process比Thread更穩定，而且Process可以分布到多臺機器上，而Thread最多只能分布到同一臺機器的多個CPU上。Python的multiprocessing模塊不但支持多進程，其中managers子模塊還支持把多進程分布到多臺機器上。一個服務進程可以作為調度者，將任務分布到其他多個進程中，依靠網絡通信。由于managers模塊封裝很好，不必了解網絡通信的細節，就可以很容易地編寫分布式多進程程序。

代碼記錄

舉個例子

如果我們已經有一個通過Queue通信的多進程程序在同一臺機器上運行，現在，由于處理任務的進程任務繁重，希望把發送任務的進程和處理任務的進程分布到兩臺機器上，這應該怎么用分布式進程來實現呢?你已經知道了原有的Queue可以繼續使用，而且通過managers模塊把Queue通過網絡暴露出去，就可以讓其他機器的進程來訪問Queue了。好，那我們就這么干!

寫個task_master.py

我們先看服務進程。服務進程負責啟動Queue，把Queue注冊到網絡上，然后往Queue里面寫入任務。

#！/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服務進程：

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進程發送完任務后，開始等待result隊列的結果。現在啟動task_worker.py進程：

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] 由于目標計算機積極拒絕，無法連接。 

看到沒，結果都出錯了，我們好好分析一下到底哪出錯了。。。

錯誤分析

在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進程中：

而Queue之所以能通過網絡訪問，就是通過QueueManager實現的。由于QueueManager管理的不止一個Queue，所以，要給每個Queue的網絡調用接口起個名字，比如get_task_queue。task_worker這里的QueueManager注冊的名字必須和task_manager中的一樣。對比上面的例子，可以看出Queue對象從另一個進程通過網絡傳遞了過來。只不過這里的傳遞和網絡通信由QueueManager完成。

authkey有什么用?這是為了保證兩臺機器正常通信，不被其他機器惡意干擾。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定連接不上。