Python分布式进程中你会遇到的坑(2)

现在，可以试试分布式进程的工作效果了。先启动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() 

（编辑：ASP站长网）