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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
class Job:
def __init__(self, job_id):
self.job_id = job_id

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()

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()