Python使用multiprocessing实现一个最简单的分布式作业调度系统_Python

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实现一个最简单的分布式作业调度系统，希望对大家有所帮助！

Python使用multiprocessing实现一个最简单的分布式作业调度系统

相关文章

热门资讯