脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 脚本专栏 → python → python 分布式调度

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

 更新时间：2016年03月14日 10:02:25   作者：kongxx  
mutilprocess像线程一样管理进程，这个是mutilprocess的核心，他与threading很是相像，对多核CPU的利用率会比threading好的多,通过本文给大家介绍Python使用multiprocessing实现一个最简单的分布式作业调度系统，需要的朋友参考下

 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中Librosa的mfcc步骤详解

    对python中Librosa的mfcc步骤详解

    今天小编就为大家分享一篇对python中Librosa的mfcc步骤详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2019-01-01
  • 基于Python编写一个语音合成系统

    基于Python编写一个语音合成系统

    这篇文章主要介绍了如何利用Python制作一个语音合成系统，文中的示例代码讲解详细，对我们学习Python有一定的帮助，感兴趣的小伙伴可以了解一下
    2022-03-03
  • win7上tensorflow2.2.0安装成功 引用DLL load failed时找不到指定模块 tensorflow has no attribute xxx 解决方法

    win7上tensorflow2.2.0安装成功 引用DLL load failed时找不到指定模块 tensorflo

    这篇文章主要介绍了win7上tensorflow2.2.0安装成功 引用时DLL load failed找不到指定模块 tensorflow has no attribute xxx 解决方法,需要的朋友可以参考下
    2020-05-05
  • python中id函数运行方式

    python中id函数运行方式

    在本篇文章里小编给大家分享了关于python中id函数运行方式及相关实例，需要的朋友们可以参考下。
    2020-07-07
  • Django Admin后台模型列表页面如何添加自定义操作按钮

    Django Admin后台模型列表页面如何添加自定义操作按钮

    这篇文章主要介绍了Django Admin后台模型列表页面如何添加自定义操作按钮,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-11-11
  • python如何生成各种随机分布图

    python如何生成各种随机分布图

    这篇文章主要为大家详细介绍了python如何生成各种随机分布图，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2018-08-08
  • python调用matlab的方法详解

    python调用matlab的方法详解

    这篇文章主要为大家介绍了python调用matlab，具有一定的参考价值，感兴趣的小伙伴们可以参考一下，希望能够给你带来帮助
    2021-11-11
  • pytorch中的squeeze函数、cat函数使用

    pytorch中的squeeze函数、cat函数使用

    这篇文章主要介绍了pytorch中的squeeze函数、cat函数使用，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-05-05
  • python中的格式化输出用法总结

    python中的格式化输出用法总结

    这篇文章主要介绍了python中的格式化输出用法,分析了Python格式化输出的种类并结合实例形式总结了针对浮点数的格式化输出方法,需要的朋友可以参考下
    2016-07-07
  • Matlab实现图像边缘检测

    Matlab实现图像边缘检测

    这篇文章主要为大家详细介绍了Matlab实现图像边缘检测，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2021-10-10

最新评论