Python使用multiprocessing实现一个最简单的分布式作业调度系统
作者:kongxx 时间:2022-06-14 07:43:33
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,分布式,调度
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
Python实现把回车符\\r\\n转换成\\n
2022-09-21 07:22:14
python检测某个变量是否有定义的方法
2021-05-17 18:45:01
Python+selenium实现自动循环扔QQ邮箱漂流瓶
2021-07-12 23:46:28
从MySQL4.0向MySQL5迁移数据
2007-11-19 13:11:00
![](https://img.aspxhome.com/file/UploadPic/200711/19/20071119131517223s.png)
使用Python标准库中的wave模块绘制乐谱的简单教程
2023-11-20 14:21:35
![](https://img.aspxhome.com/file/2023/5/63525_0s.jpg)
JavaScript解决Joseph问题
2008-06-21 17:11:00
php curl登陆qq后获取用户信息时证书错误
2023-11-15 08:23:39
golang 网络框架之gin的使用方法
2023-07-19 02:35:37
ASP利用TCPIP.DNS组件获得域名对应的IP
2009-11-07 19:21:00
ie7.0浏览器 兼容问题苦煞网站设计者
2007-08-08 17:11:00
Python爬虫爬取微信朋友圈
2021-11-12 17:38:14
![](https://img.aspxhome.com/file/2023/1/76811_0s.jpg)
pytest多重断言的实现
2021-10-12 03:30:25
PDO::_construct讲解
2023-06-06 03:22:31
python使用matplotlib显示图像失真的解决方案
2021-03-30 22:31:02
![](https://img.aspxhome.com/file/2023/6/73006_0s.jpg)
使用mysqli完成事务处理
2011-03-29 15:49:00
WEB2.0网页制作标准教程(5)head区的其他设置
2007-11-13 13:28:00
pytest内置fixture使用临时目录流程详解
2021-12-27 06:49:23
ASP截取中英文字符串固定长度
2009-08-19 17:12:00
使用DIV+CSS设计网页的好处
2007-10-14 15:02:00
Python基础之getpass模块详细介绍
2021-03-06 13:47:13
![](https://img.aspxhome.com/file/2023/3/66793_0s.jpg)