Python multiprocessing.managers实现分布式多进程详解
本文详解Python中利用multiprocessing模块的managers子模块实现分布式多进程的方法。通过QueueManager将任务队列暴露至网络，构建Master/Worker模型，实现跨机器调度。文章涵盖服务进程与任务进程的代码实现、网络配置、authkey验证机制及性能优化建议，帮助开发者轻松搭建高效的分布式计算环境。
在多线程与多进程之间，推荐优先选用Process。这是因为Process运行更为稳定，且具备分布到多台机器上的能力；相比之下，Thread最多只能分布在同一台机器的多个CPU核心上。
Python的multiprocessing模块不仅支持单机多进程，其managers子模块进一步支持将多进程分布到多台机器上。通过一个服务进程作为调度中心，利用网络通信将任务分发至其他多个进程。由于managers模块封装完善，开发者无需深入理解网络通信细节，即可便捷地编写分布式多进程程序。
举例说明：假设我们已有一个基于Queue通信的多进程程序在同一台机器上运行。若处理任务的进程负载过重，希望将发送任务的进程与处理任务的进程分离至两台不同的机器上，该如何实现分布式进程架构？
原有的Queue对象可以继续使用，但关键在于通过managers模块将Queue暴露到网络上，从而使其他机器上的进程能够访问该Queue。
首先查看服务进程的代码实现。服务进程负责启动Queue，将其注册到网络中，并向Queue内写入任务：
# 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()
需要注意的是，在单机多进程程序中，创建的Queue可直接使用；但在分布式多进程环境下，向Queue添加任务时不可直接操作原始的task_queue，因为这会绕过QueueManager的封装。必须通过manager.get_task_queue()获取的Queue接口来添加任务。
接下来，在另一台机器上（或本机）启动任务进程：
# 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()
任务进程需要通过网络连接到服务进程，因此必须指定服务进程的IP地址。
现在，我们可以测试分布式进程的工作效果。首先启动task_master.py服务进程：
$ python3 task_master.py Put task 3411... Put task 1605... Put task 1398... Put task 4729... Put task 5300... Put task 7471... Put task 68... Put task 4219... Put task 339... Put task 7866... Try get results...
task_master.py进程在发送完所有任务后，会开始等待result队列返回结果。此时启动task_worker.py进程：
$ python3 task_worker.pyConnect to server 127.0.0.1... run task 3411 * 3411... run task 1605 * 1605... run task 1398 * 1398... run task 4729 * 4729... run task 5300 * 5300... run task 7471 * 7471... run task 68 * 68... run task 4219 * 4219... run task 339 * 339... run task 7866 * 7866... worker exit.
当task_worker.py进程执行完毕后，在task_master.py进程中会继续打印出计算结果：
Result: 3411 * 3411 = 11634921 Result: 1605 * 1605 = 2576025 Result: 1398 * 1398 = 1954404 Result: 4729 * 4729 = 22363441 Result: 5300 * 5300 = 28090000 Result: 7471 * 7471 = 55815841 Result: 68 * 68 = 4624 Result: 4219 * 4219 = 17799961 Result: 339 * 339 = 114921 Result: 7866 * 7866 = 61873956
这种简单的Master/Worker模型有何实际用途？这实际上是一个真实且基础的分布式计算架构。只需对代码稍作调整，启动多个worker进程，即可将任务分布到几台甚至几十台机器上。例如，若将计算n*n的代码替换为发送邮件的代码，即可实现邮件队列的异步发送功能。
Queue之所以能通过网络被访问，是通过QueueManager实现的。由于QueueManager管理着不止一个Queue，因此需要为每个Queue的网络调用接口指定名称，例如get_task_queue。
authkey的作用是什么？这是为了确保两台机器间的正常通信，防止被其他恶意机器干扰。如果task_worker.py中的authkey与task_master.py中的authkey不一致，连接将无法建立。
Python的分布式进程接口设计简单且封装良好，非常适合需要将繁重任务分布到多台机器的应用场景。
需注意，Queue的主要作用是传递任务描述和接收结果，因此每个任务的描述数据量应尽可能小。例如，若需处理日志文件，不应发送几百兆的日志文件本身，而应发送日志文件存放的完整路径，由Worker进程再去共享磁盘上读取文件。
以上内容即为本文所介绍的全部知识。本文主要讲解了Python分布式进程的核心概念与实现方法，希望读者能通过这些资料深入理解相关内容。期望本文内容能对您有所帮助，使您的Python学习过程更加轻松顺畅。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
