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 queue
from multiprocessing.managers import basemanager
from 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_queue
def get_finished_job_queue(self):
return self.finished_job_queue
def 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 = 0
while true:
for i in range(0, 10):
job_id = job_id + 1
job = 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 time
from queue import queue
from multiprocessing.managers import basemanager
from 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’)
# 连接master
server = ‘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: 1
dispatch job: 2
dispatch job: 3
dispatch job: 4
dispatch job: 5
dispatch job: 6
dispatch job: 7
dispatch job: 8
dispatch job: 9
dispatch job: 10
finished job: 1
finished job: 2
finished job: 3
finished job: 4
finished job: 5
finished job: 6
finished job: 7
finished job: 8
finished job: 9
dispatch job: 11
dispatch job: 12
dispatch job: 13
dispatch job: 14
dispatch job: 15
dispatch job: 16
dispatch job: 17
dispatch job: 18
dispatch job: 19
dispatch job: 20
finished job: 10
finished job: 11
finished job: 12
finished job: 13
finished job: 14
finished job: 15
finished job: 16
finished job: 17
finished job: 18
dispatch job: 21
dispatch job: 22
dispatch job: 23
dispatch job: 24
dispatch job: 25
dispatch job: 26
dispatch job: 27
dispatch job: 28
dispatch job: 29
dispatch 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实现一个最简单的分布式作业调度系统，希望对大家有所帮助！