- 当前位置：首页 >应用开发 >Python分布式进程中你会遇到的坑
Python分布式进程中你会遇到的坑
发布时间：2025-11-05 15:31:54 来源：创站工坊 作者：系统运维
小惊大怪
你是分布不是在用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=babc) # 启动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=babc) # 从网络连接: 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: Cant pickle <function <lambda> at 0x00000202D1921E18>: attribute lookup <lambda> on __main__ failedtask_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=babc)修改后的代码
在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=babc) # windows需要写ip地址 manager = QueueManager(address=(127.0.0.1, 5000), authkey=babc) 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=babc) # 从网络连接: 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不一致，肯定连接不上。
- 惠普电脑,预装ubuntu系统如何修改系统语言1、首先，我们打开Ubuntu桌面左侧工具栏中的”系统设置”2、然后选中设置中的“Language Support”3、这里的默认系统语言为English4、我们找到汉语（中国），向上拉，使汉语（中国）的位置替换English的位置。有的时候，不好拉，多拉几次就好了5、将汉语（中国）拉到顶部后，我们点击“Apply System-Wide”,将汉语（中国）应用到整个系统。6、在弹出来的对话框中，输入你的root密码之后，点击“Authenticate”。等一会儿，待应用完成之后，我们重启7、在重启之后，我们会看到，有一个询问的对话框，问你是否将标准文件夹更新到当前语言，这里按照个人喜好而定”。我们这里也已经看到，系统语言已经由English修改为了汉语。至此，完成了系统语言的更换
- 微软Windows7桌面三大个性化功能
- 重装Win 7系统导入XP模式等虚拟机重装系统方法
- windows 7开机提示系统自动修复无法正常进入的原因分析及解决
- 卧龙大数据（发挥数据的力量，卧龙大数据引领行业变革）
- 优化让windows7开机加速更快点!
- 如何开启windows 7端口的图文教程
- windows 7系统下卸载.net framework的方法图文教程
- 虚拟机下面安装了ubuntu系统，显示的屏幕只有那么一小块儿，不知道如何才能全屏，那么如何全屏呢？且看下面经验。1、打开虚拟机，并点击要更改成全屏的那个ubuntu系统的电源，我的虚拟机名字就叫ubuntu，那么就点击【打开此虚拟机电源】2、等虚拟机打开之后，我们点击虚拟机软件上面工具栏中的【虚拟机(V)】，会展现出一个下拉菜单。3、在下拉菜单中，我们找到并使用鼠标左键单击【安装Vmware工具】，假如你这里是灰色的，那么可能是 因为你的虚拟机版本比较低！4、点击以上选项后，我们进入到系统里面，找到在桌面上出现的wmware tools的光盘！我们点击进入其中。5、在vmware tools虚拟光盘里面，我们双击【vmware****.tar.gz】这个文件，注意我这里的****是任意字符的意思哦，每个虚拟机的版本可能不一。6、复制这个【vmware****.tar.gz】文件，到【文件】--->【home】文件夹里面。7、然后按【Ctrl+Alt+T】调出命令界面，然后在里面输入【tar -zxvf v】后按【Tab】键，自动补全整个工具的名字。然后按enter执行。8、然后在终端里面输入【cd V】，再按一次TAB键补全被解压后的那个工具目录名字。回车后进入到该工具解压后的目录里面！最后输入【sudo ./vmware-install.pl】执行即可安装成功，安装成功后，按【CTRL+Alt+enter】键就能给ubuntu全屏啦。假如还有疑问，可以提问我，我会第一时间回答的。
- Windows7任务栏桌面下角的一些正在运行的图标不见了
- Copyright © 2025 Powered by Python分布式进程中你会遇到的坑,创站工坊 滇ICP备2023006006号-44sitemap