Skip to main content
 Web开发网 » 编程语言 » Python语言

如何使用Python的multiprocessing进行分布式计算

2021年11月27日520百度已收录

如何使用Python的multiprocessing进行分布式计算  Python分布式计算 第1张

由于GIL的存在,python中的多线程其实并不是真正意义上的多线程,前面一篇文章Python中的多线程与多进程那些事 中提到

I/O密集型使用多线程并发执行提高效率、计算密集型使用多进程并行执行提高效率。针对计算密集型的任务,我们如何通过多进程提高效率?

Window下的multiprocessing分布式计算基于multiprocessing、queue等模块实现简易的分布式计算框架。服务节点负责任务的派发和任务结果的采集,工作节点分布在同一电脑的不同进程,或者其他电脑上,负责任务执行和结果反馈,服务节点与工作节点通过Queue实现数据共享(任务下发、结果反馈)。

其中包含Task、MaterWork、SlaveWork三个模块,大体介绍及源码如下:

Task模块根据任务需求,将自己的任务单独封装在task模块中。

#!/usr/bin/env python

# -*- coding: utf-8 -*-

class Task:

def __init__(self, job_id):

self.task_id = job_id

def computer(self):

return self.task_id*self.task_id

MaterWork(主节点/服务节点)负责任务派发和结果采集,主从节点通过Queue实现任务、结果共享。代码中注释中已介绍的较清楚,这里不做赘述。

#!/usr/bin/env python

# _*_ coding:utf-8 _*_

import Queue

from multiprocessing.managers import BaseManager

from Task import Task

#window下运行防止多进程可能崩溃

from multiprocessing import freeze_support

class QueueManager(BaseManager):

pass

# 发送任务的队列

task_queue = Queue.Queue()

# 接收结果的队列

result_queue = Queue.Queue()

# 解决__main__.<lambda> not found问题

def get_task_queue():

return task_queue

# 解决__main__.<lambda> not found问题

def get_result_queue():

return result_queue

def Master_Work():

# 把两个queue注册到网络上

QueueManager.register('get_dispatched_job_queue', callable=get_task_queue)

QueueManager.register('get_finished_job_queue', callable=get_result_queue)

# 绑定端口9527,设置验证码123456

manager = QueueManager(address=('192.168.48.89', 9527), authkey='123456')

manager.start()

# 通过网络访问Queue对象

dispatched_jobs = manager.get_dispatched_job_queue()

finished_jobs = manager.get_finished_job_queue()

# 添加待处理任务,实际应用过程添加的任务可以为函数的参数

for i in range(0, 10):

task = Task(i).computer()

print 'Dispatch job: {0}'.format(i)

dispatched_jobs.put(task)

# 列队任务消费完,开始获取任务执行结果

while not dispatched_jobs.empty():

for i in range(0, 10):

TaskResult = finished_jobs.get(60)

print 'Finished Job {0}'.format(TaskResult)

# 关闭服务

manager.shutdown()

if __name__ == '__main__':

freeze_support()

Master_Work()

SlaveWork(从节点/计算节点) 负责任务执行和结果反馈,主从节点通过Queue实现任务、结果共享。

#!/usr/bin/env python

# _*_ coding:utf-8 _*_

import Queue

from multiprocessing.managers import BaseManager

import time

class QueueManager(BaseManager):

pass

# 从网络上获取Queue

QueueManager.register('get_dispatched_job_queue')

QueueManager.register('get_finished_job_queue')

# 连接服务器

server_addr = '192.168.48.89'

print 'Connect to server {0}'.format(server_addr)

manager = QueueManager(address=(server_addr, 9527), authkey='123456')

manager.connect()

# 获取Queue对象

dispatched_jobs = manager.get_dispatched_job_queue()

finished_jobs = manager.get_finished_job_queue()

# 消费队列执行任务

while not dispatched_jobs.empty():

try:

#任务函数 此处可添加任务函数

task = dispatched_jobs.get(timeout=1)

print('SlaveWork1 Run job: {0}'.format(task) )

finished_jobs.put(task)

except Queue.Empty:

print 'task queue is empty'

print 'worker exit!'

使用说明3.1 在Task.py 添加任务函数。

如何使用Python的multiprocessing进行分布式计算  Python分布式计算 第2张

一个小栗子

3.2 调整主从节点任务派发具体函数、次数或参数。

一般情况分发次数等于任务数,即分发任务的参数。

# 添加待处理任务,实际应用过程添加的任务可以为函数的参数

for i in range(0, 10):

task = Task(i).computer()

print 'Dispatch job: {0}'.format(i)

dispatched_jobs.put(task)

3.3 启动主节点

如何使用Python的multiprocessing进行分布式计算  Python分布式计算 第3张

3.3 启动从节点

如何使用Python的multiprocessing进行分布式计算  Python分布式计算 第4张

将从节点计算模块发送至不同电脑,启动从节点即实现分布式计算,即N个从节点消费服务节点分发任务,并返回任务执行状态、结果。

3.4 启动从节点后,可以在主节点运行界面,看到任务完成情况。

如何使用Python的multiprocessing进行分布式计算  Python分布式计算 第5张

转载请说明哦,若对你有帮助,点赞支持一下。

评论列表暂无评论
发表评论
微信