Python分布式多进程计算实战：基于BaseManager的架构解析

Python分布式多进程计算实战：基于BaseManager的架构解析

前言

分布式多进程的优势与适用场景

在Python中，多进程和多线程均可用于并发执行多个任务。在两者之间进行选择时，通常优先推荐多进程方案，因其运行更为稳定，且进程间的资源隔离使得管理更为便捷。然而，多进程有一个独特的优势：支持将进程分布到多台计算机上运行。当面临海量任务且单机即使启用多进程仍耗时过长时，将任务分散至多台机器并行处理成为提升效率的关键策略。

基于BaseManager的通信机制与流程

在分布式进程运算架构中，进程间通信依然依赖Queue队列，但此时不能直接使用标准队列对象，而必须使用multiprocessing.managers.BaseManager 对其进行包装。只有经过封装并通过回调机制处理后，队列才能在网络环境中使用。分布式调用架构通常包含服务端和客户端两部分：服务端通过网络协议将队列中的任务分发给各个客户端执行，客户端计算完成后将结果回传至队列，最终由服务端统一收集并展示结果。具体流程如下所示。

分布式流程示意图

服务端实现细节与代码解析

服务端将任务放入task_queue队列中，随后四个客户端通过网络端口从该队列获取任务进行计算，并将结果存入result_queue队列，最后由服务端统一处理并输出结果。整体流程逻辑清晰，但需特别注意：此处的队列不能是原始的queue.Queue对象，必须通过BaseManager进行封装处理。

首先分析服务端的实现代码。

#coding:gbk
import time, queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support

# 任务个数
task_number = 10
# 定义收发队列
task_queue = queue.Queue(task_number)
result_queue = queue.Queue(task_number)


def gettask():
 return task_queue


def getresult():
 return result_queue


def test():
 # windows下绑定调用接口不能使用lambda，所以只能先定义函数再绑定
 BaseManager.register('get_task', callable=gettask)
 BaseManager.register('get_result', callable=getresult)
 # 绑定端口并设置验证码，windows下需要填写ip地址，linux下不填默认为本地
 manager = BaseManager(address=('127.0.0.1', 5002), authkey=b'123')
 # 启动
 manager.start()
 try:
  # 通过网络获取任务队列和结果队列
  task = manager.get_task()
  result = manager.get_result()
  # 添加任务
  for i in range(task_number):
   print('Put task %d...' % i)
   task.put(i)
  # 每秒检测一次是否所有任务都被执行完
  while not result.full():
   print(task.qsize())
   time.sleep(1)
  for i in range(result.qsize()):
   ans = result.get()
   print('task %d is finish , runtime:%d s' % ans)
 except:
  print('Manager error')
 finally:
  manager.shutdown()


if __name__ == '__main__':
 # windows下多进程可能会炸，添加这句可以缓解
 freeze_support()
 test()

此处重点解析BaseManager.register('get_task', callable=gettask) 这行代码，其功能是注册一个名为get_task的操作，该操作实际执行的是gettask()函数。在前文定义中，gettask()函数负责返回task_queue。这正是前文所述不能直接使用queue.Queue的原因，必须通过BaseManager的register接口封装后的队列才能通过网络传输。随后，代码使用task = manager.get_task()获取封装后的队列实例。

manager = BaseManager(address=('127.0.0.1', 5002), authkey=b'123')

该行代码初始化了一个Manager实例，它绑定了本机5002端口，并在客户端连接时要求提供密码：123。

客户端配置与运行测试

接下来查看客户端的代码实现。

#coding:gbk

import time, sys, queue, random
from multiprocessing.managers import BaseManager
BaseManager.register('get_task')
BaseManager.register('get_result')
conn = BaseManager(address = ('127.0.0.1',5002), authkey = b'123')
try:
 conn.connect()
except:
 print('连接失败')
 sys.exit()
task = conn.get_task()
result = conn.get_result()
while not task.empty():
 print(task.qsize())
 n = task.get(timeout = 1)
 print('run task %d' % n)
 sleeptime = random.randint(0,3)
 time.sleep(sleeptime)
 rt = (n, sleeptime)
 result.put(rt)
if __name__ == '__main__':
 pass;

主要关注以下代码片段。

BaseManager.register('get_task')
BaseManager.register('get_result')

这两行代码用于注册函数，需与服务端注册的函数保持一致。由于服务端已注册了这两个函数，客户端才能调用它们。需要注意的是，客户端不能注册服务端未注册的函数。

运行测试时，先启动服务端程序，随后开启两个CMD窗口运行客户端程序。此外，也可以在局域网内的其他机器上运行客户端，但需相应修改服务端代码中的IP地址以匹配网络环境。

服务端的运行结果如下所示。

Put task 0... Put task 1... Put task 2... Put task 3... Put task 4... Put task 5... Put task 6... Put task 7... Put task 8... Put task 9... task 0 is finish , runtime:3 s task 1 is finish , runtime:0 s task 2 is finish , runtime:2 s task 4 is finish , runtime:1 s task 3 is finish , runtime:3 s task 6 is finish , runtime:1 s task 7 is finish , runtime:0 s task 5 is finish , runtime:3 s task 8 is finish , runtime:2 s task 9 is finish , runtime:3 s

两个客户端的运行结果分别如下。

客户端1的输出日志

10 run task 0 9 run task 1 8 run task 2 6 run task 4 5 run task 5 1 run task 9

客户端2的输出日志

7 run task 3 4 run task 6 3 run task 7 2 run task 8

多程序共同运行的截图展示。

运行状态截图

由于Queue本身是线程安全的，因此此处无需额外添加锁机制。在客户端代码中打印print(task.qsize())以查看当前队列大小，可以观察到队列信息已同步至各个客户端。

生产环境建议与总结

最后需要补充说明的是，虽然分布式多进程能够将任务分散到不同机器上运行，从而处理大规模并发任务，但如果服务端发生故障，所有任务将会丢失。因此，在生产环境中，建议考虑使用如Kafka等成熟的消息中间件来保障系统的可靠性。

总结

以上为本文的全部内容，希望本文的介绍对您的学习或工作具有参考价值。感谢大家对本站的支持。