Python multiprocessing分布式进程实战：解决Windows序列化与连接错误

清欢

发布时间：2026-10-13 | 4人浏览过

原创

本文详解Python中利用multiprocessing模块实现分布式进程的方法。重点解析在Windows环境下使用QueueManager注册网络队列时，因lambda匿名函数无法序列化导致的PicklingError，以及连接被拒绝的常见错误。通过对比修改前后的代码，展示如何正确封装Queue对象、配置authkey及处理网络通信细节，帮助开发者避开分布式计算中的典型坑点。

Python分布式进程概述

小惊大怪

如果你正在使用Python 3或Windows系统进行编程，且对进程与线程的概念尚不熟悉，那么在尝试构建Python分布式进程时可能会遇到一些挑战。不过别担心，只要了解序列化处理中不支持匿名函数这一关键点，就能避开大部分陷阱。接下来我们将直接进入核心内容。

分布式进程

相较于线程，进程具有更高的稳定性，并且能够分布到多台机器上运行，而线程通常仅限于同一台机器的多个CPU核心。Python的multiprocessing模块不仅支持多进程，其子模块managers更是支持将多进程分布到多台机器。通过一个服务进程作为调度者，利用网络通信将任务分发到其他进程，managers模块封装了网络通信细节，使得编写分布式多进程程序变得简单。

代码记录

分布式进程实现示例

举个例子

假设你已有一个通过Queue通信的多进程程序在同一台机器上运行。若处理任务的进程负载过重，希望将发送任务的进程和处理任务的进程分布到两台不同的机器上，该如何实现？原有的Queue可以继续使用，只需通过managers模块将Queue暴露到网络中，其他机器的进程即可访问该Queue。

写个task_master.py

首先编写服务进程代码。该进程负责启动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函数上。这是因为Python的序列化机制不支持匿名函数。因此，我们需要修改代码，重新对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中的数据需要等待Worker机器运算处理后才能读取。因此，需要通过上述两行代码，利用QueueManager将Queue封装并暴露到网络中。我们为return_task_queue的网络调用接口命名为get_task_queue，为return_result_queue命名为get_result_queue，以便区分操作对象。task.put(n)即向task_queue写入数据，相当于分配任务；而result.get()则是等待Worker处理返回的结果。

值得注意的是，在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不一致，连接将无法建立。

总结

以上所述是Python分布式进程中常见问题的解析，希望对大家有所帮助。如有任何疑问，请留言，小编会及时回复。感谢大家对本站的支持！如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！