mutilprocess像线程一样管理进程，这个是mutilprocess的核心，他与threading很是相像，对多核CPU的利用率会比threading好的多。
介绍
Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程分布到多台机器上。一个服务进程可以作为调度者，将任务分布到其他多个机器的多个进程中，依靠网络通信。
想到这，就在想是不是可以使用此模块来实现一个简单的作业调度系统。
实现
Job
首先创建一个Job类，为了测试简单，只包含一个job id属性
job.py
#!/usr/bin/env python# -*- coding: utf-8 -*-class Job:def __init__(self, job_id):self.job_id = job_id
Master
Master用来派发作业和显示运行完成的作业信息
master.py
#!/usr/bin/env python# -*- coding: utf-8 -*-from Queue import Queuefrom multiprocessing.managers import BaseManagerfrom job import Job
class Master:
def __init__(self):# 派发出去的作业队列self.dispatched_job_queue = Queue()# 完成的作业队列self.finished_job_queue = Queue()def get_dispatched_job_queue(self):return self.dispatched_job_queuedef get_finished_job_queue(self):return self.finished_job_queuedef start(self):# 把派发作业队列和完成作业队列注册到网络上BaseManager.register('get_dispatched_job_queue', callable=self.get_dispatched_job_queue)BaseManager.register('get_finished_job_queue', callable=self.get_finished_job_queue)# 监听端口和启动服务manager = BaseManager(address=('0.0.0.0', 8888), authkey='jobs')manager.start()# 使用上面注册的方法获取队列dispatched_jobs = manager.get_dispatched_job_queue()finished_jobs = manager.get_finished_job_queue()# 这里一次派发10个作业，等到10个作业都运行完后，继续再派发10个作业job_id = 0while True:for i in range(0, 10):job_id = job_id + 1job = Job(job_id)print('Dispatch job: %s' % job.job_id)dispatched_jobs.put(job)while not dispatched_jobs.empty():job = finished_jobs.get(60)print('Finished Job: %s' % job.job_id)manager.shutdown()if __name__ == "__main__":master = Master()master.start() Slave
Slave用来运行master派发的作业并将结果返回
slave.py
#!/usr/bin/env python# -*- coding: utf-8 -*-import timefrom Queue import Queuefrom multiprocessing.managers import BaseManagerfrom job import Job
class Slave:
def __init__(self):# 派发出去的作业队列self.dispatched_job_queue = Queue()# 完成的作业队列self.finished_job_queue = Queue()
def start(self):
# 把派发作业队列和完成作业队列注册到网络上BaseManager.register('get_dispatched_job_queue')BaseManager.register('get_finished_job_queue')# 连接masterserver = '127.0.0.1'print('Connect to server %s...' % server)manager = BaseManager(address=(server, 8888), authkey='jobs')manager.connect()# 使用上面注册的方法获取队列dispatched_jobs = manager.get_dispatched_job_queue()finished_jobs = manager.get_finished_job_queue()# 运行作业并返回结果，这里只是模拟作业运行，所以返回的是接收到的作业while True:job = dispatched_jobs.get(timeout=1)print('Run job: %s ' % job.job_id)time.sleep(1)finished_jobs.put(job)if __name__ == "__main__":slave = Slave()slave.start() 测试
分别打开三个linux终端，第一个终端运行master，第二个和第三个终端用了运行slave，运行结果如下
master
$ python master.py Dispatch job: 1Dispatch job: 2Dispatch job: 3Dispatch job: 4Dispatch job: 5Dispatch job: 6Dispatch job: 7Dispatch job: 8Dispatch job: 9Dispatch job: 10Finished Job: 1Finished Job: 2Finished Job: 3Finished Job: 4Finished Job: 5Finished Job: 6Finished Job: 7Finished Job: 8Finished Job: 9Dispatch job: 11Dispatch job: 12Dispatch job: 13Dispatch job: 14Dispatch job: 15Dispatch job: 16Dispatch job: 17Dispatch job: 18Dispatch job: 19Dispatch job: 20Finished Job: 10Finished Job: 11Finished Job: 12Finished Job: 13Finished Job: 14Finished Job: 15Finished Job: 16Finished Job: 17Finished Job: 18Dispatch job: 21Dispatch job: 22Dispatch job: 23Dispatch job: 24Dispatch job: 25Dispatch job: 26Dispatch job: 27Dispatch job: 28Dispatch job: 29Dispatch job: 30
slave1
$ python slave.py Connect to server 127.0.0.1...Run job: 1 Run job: 2 Run job: 3 Run job: 5 Run job: 7 Run job: 9 Run job: 11 Run job: 13 Run job: 15 Run job: 17 Run job: 19 Run job: 21 Run job: 23
slave2
$ python slave.py Connect to server 127.0.0.1...Run job: 4 Run job: 6 Run job: 8 Run job: 10 Run job: 12 Run job: 14 Run job: 16 Run job: 18 Run job: 20 Run job: 22 Run job: 24
以上内容是小编给大家介绍的Python使用multiprocessing实现一个最简单的分布式作业调度系统，希望对大家有所帮助！
