快捷導(dǎo)航

Python使用multiprocessing實(shí)現(xiàn)一個(gè)最簡(jiǎn)單的分布式作業(yè)調(diào)度系統(tǒng)

更新時(shí)間：2016年03月14日 10:02:25 作者：kongxx

mutilprocess像線程一樣管理進(jìn)程，這個(gè)是mutilprocess的核心，他與threading很是相像，對(duì)多核CPU的利用率會(huì)比threading好的多,通過(guò)本文給大家介紹Python使用multiprocessing實(shí)現(xiàn)一個(gè)最簡(jiǎn)單的分布式作業(yè)調(diào)度系統(tǒng)，需要的朋友參考下

mutilprocess像線程一樣管理進(jìn)程，這個(gè)是mutilprocess的核心，他與threading很是相像，對(duì)多核CPU的利用率會(huì)比threading好的多。

介紹

Python的multiprocessing模塊不但支持多進(jìn)程，其中managers子模塊還支持把多進(jìn)程分布到多臺(tái)機(jī)器上。一個(gè)服務(wù)進(jìn)程可以作為調(diào)度者，將任務(wù)分布到其他多個(gè)機(jī)器的多個(gè)進(jìn)程中，依靠網(wǎng)絡(luò)通信。

想到這，就在想是不是可以使用此模塊來(lái)實(shí)現(xiàn)一個(gè)簡(jiǎn)單的作業(yè)調(diào)度系統(tǒng)。

實(shí)現(xiàn)

Job

首先創(chuàng)建一個(gè)Job類，為了測(cè)試簡(jiǎn)單，只包含一個(gè)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用來(lái)派發(fā)作業(yè)和顯示運(yùn)行完成的作業(yè)信息

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):
# 派發(fā)出去的作業(yè)隊(duì)列
self.dispatched_job_queue = Queue()
# 完成的作業(yè)隊(duì)列
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):
# 把派發(fā)作業(yè)隊(duì)列和完成作業(yè)隊(duì)列注冊(cè)到網(wǎng)絡(luò)上
BaseManager.register('get_dispatched_job_queue', callable=self.get_dispatched_job_queue)
BaseManager.register('get_finished_job_queue', callable=self.get_finished_job_queue)
# 監(jiān)聽(tīng)端口和啟動(dòng)服務(wù)
manager = BaseManager(address=('0.0.0.0', 8888), authkey='jobs')
manager.start()
# 使用上面注冊(cè)的方法獲取隊(duì)列
dispatched_jobs = manager.get_dispatched_job_queue()
finished_jobs = manager.get_finished_job_queue()
# 這里一次派發(fā)10個(gè)作業(yè)，等到10個(gè)作業(yè)都運(yùn)行完后，繼續(xù)再派發(fā)10個(gè)作業(yè)
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用來(lái)運(yùn)行master派發(fā)的作業(yè)并將結(jié)果返回

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):
# 派發(fā)出去的作業(yè)隊(duì)列
self.dispatched_job_queue = Queue()
# 完成的作業(yè)隊(duì)列
self.finished_job_queue = Queue()

def start(self):

# 把派發(fā)作業(yè)隊(duì)列和完成作業(yè)隊(duì)列注冊(cè)到網(wǎng)絡(luò)上
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()
# 使用上面注冊(cè)的方法獲取隊(duì)列
dispatched_jobs = manager.get_dispatched_job_queue()
finished_jobs = manager.get_finished_job_queue()
# 運(yùn)行作業(yè)并返回結(jié)果，這里只是模擬作業(yè)運(yùn)行，所以返回的是接收到的作業(yè)
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()

測(cè)試

分別打開(kāi)三個(gè)linux終端，第一個(gè)終端運(yùn)行master，第二個(gè)和第三個(gè)終端用了運(yùn)行slave，運(yùn)行結(jié)果如下

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

以上內(nèi)容是小編給大家介紹的Python使用multiprocessing實(shí)現(xiàn)一個(gè)最簡(jiǎn)單的分布式作業(yè)調(diào)度系統(tǒng)，希望對(duì)大家有所幫助！

您可能感興趣的文章: