由于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 添加任务函数。
一个小栗子
3.2 调整主从节点任务派发具体函数、次数或参数。
一般情况分发次数等于任务数，即分发任务的参数。
# 添加待处理任务，实际应用过程添加的任务可以为函数的参数
for i in range(0, 10):
task = Task(i).computer()
print 'Dispatch job: {0}'.format(i)
dispatched_jobs.put(task)
3.3 启动主节点
3.3 启动从节点
将从节点计算模块发送至不同电脑，启动从节点即实现分布式计算，即N个从节点消费服务节点分发任务，并返回任务执行状态、结果。
3.4 启动从节点后，可以在主节点运行界面，看到任务完成情况。
转载请说明哦，若对你有帮助，点赞支持一下。
