Python multiprocessing实现分布式作业调度系统详解
本文详细介绍如何利用Python的multiprocessing模块构建简易分布式作业调度系统。通过BaseManager实现跨机器进程通信，结合Process和Pool类管理任务分发与执行，涵盖主从节点架构设计及代码实现，为多核CPU高效利用提供实践参考。
介绍
Python的multiprocessing模块不仅支持本地多进程处理，其子模块managers还具备将多进程分布到多台机器上的能力。通过一个服务进程作为调度中心，利用网络通信将任务分发至其他机器的多个进程中。基于这一特性，我们可以尝试构建一个简易的作业调度系统。在深入实现之前，有必要先对Python中的多进程管理包multiprocessing进行详细解析。
multiprocessing.Process
multiprocessing包是Python标准的多进程管理库。其设计思路与threading.Thread类似，通过multiprocessing.Process对象即可创建进程。该进程可以执行Python程序中定义的函数。Process对象的使用方法与Thread对象高度一致，提供了is_alive()、join([timeout])、run()、start()、terminate()等方法。其属性包括authkey、daemon（需通过start()设置）、exitcode（进程运行时为None，若为–N则表示被信号N终止）、name和pid。此外，multiprocessing包中还提供了Lock、Event、Semaphore、Condition等同步类，用法与threading包中的同名类相同。multiprocessing的大部分API与threading保持一致，只是应用场景从多线程切换到了多进程。
该模块用于像管理线程一样管理进程，这是multiprocessing的核心组件。它与threading非常相似，但在多核CPU环境下的利用率远高于threading。
查看Process类的构造方法：
__init__(self, group=None, target=None, name=None, args=(), kwargs={})
参数说明：
- group：指定进程所属组，通常不使用。
- target：指定调用的目标函数。
- args：指定调用目标函数的位置参数元组。
- name：进程别名。
- kwargs：指定调用目标函数的关键字参数字典。
创建进程的简单示例：
#coding=utf-8 import multiprocessing def do(n) : #获取当前线程的名字 name = multiprocessing.current_process().name print name,'starting' print "worker ", n return if __name__ == '__main__' : numList = [] for i in xrange(5) : p = multiprocessing.Process(target=do, args=(i,)) numList.append(p) p.start() p.join() print "Process end."
执行结果：
Process-1 starting worker 0 Process end. Process-2 starting worker 1 Process end. Process-3 starting worker 2 Process end. Process-4 starting worker 3 Process end. Process-5 starting worker 4 Process end.
在创建子进程时，只需传入一个执行函数及其参数，创建一个Process实例，并调用其start()方法启动。join()方法用于等待子进程结束后再继续运行主进程，常用于进程间的同步操作。
注意：
在Windows系统中使用进程模块时，必须将相关进程代码写在当前.py文件的if __name__ == ‘__main__' :语句下方，否则无法正常使用Windows下的进程模块。而在Unix/Linux环境下则无此限制。
multiprocess.Pool
当操作对象数量较少时，可以直接利用multiprocessing中的Process动态生成多个进程，处理十几个任务尚可接受。但如果涉及上百甚至上千个目标，手动限制进程数量会显得繁琐。此时，使用进程池（Pool）的功能更为合适。
Pool可以提供指定数量的进程供用户调用。当有新请求提交到Pool中时，如果池未满，则创建新进程执行该请求；如果池中进程数已达规定最大值，则请求会等待，直到池中有进程结束，才会创建新进程来处理该请求。
apply_async和apply
函数原型：
apply_async(func[, args=()[, kwds={}[, callback=None]]])
两者都用于向进程池添加新进程。不同之处在于，apply每次添加新进程时，主进程与新进程并行执行，但主进程会阻塞，直到新进程的函数执行结束。这种方式效率较低，因此在Python 3.x之后不再推荐使用。
apply_async和apply功能相同，但主进程不会阻塞。
# -*- coding:utf-8 -*-
import multiprocessing
import time
def func(msg):
print "*msg: ", msg
time.sleep(3)
print "*end"
if __name__ == "__main__":
# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
pool = multiprocessing.Pool(processes=3)
for i in range(10):
msg = "hello [{}]".format(i)
# pool.apply(func, (msg,))
pool.apply_async(func, (msg,)) # 异步开启进程, 非阻塞型, 能够向池中添加进程而不等待其执行完毕就能再次执行循环
print "--" * 10
pool.close() # 关闭pool, 则不会有新的进程添加进去
pool.join() # 必须在join之前close, 然后join等待pool中所有的线程执行完毕
print "All process done."
运行结果：
"D:\Program Files\Anaconda2\python.exe" E:/pycharm/test/multiprocessing/v1.py -------------------- *msg: hello [0] *msg: hello [1] *msg: hello [2] *end *msg: hello [3] *end *end *msg: hello [4] *msg: hello [5] *end *msg: hello [6] *end *end *msg: hello [7] *msg: hello [8] *end *msg: hello [9] *end*end *end All process done. Process finished with exit code 0
获取进程的执行结果
# -*- coding:utf-8 -*-
import multiprocessing
import time
def func_with_return(msg):
print "*msg: ", msg
time.sleep(3)
print "*end"
return "{} return".format(msg)
if __name__ == "__main__":
# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
pool = multiprocessing.Pool(processes=3)
results = []
for i in range(10):
msg = "hello [{}]".format(i)
res = pool.apply_async(func_with_return, (msg,)) # 异步开启进程, 非阻塞型, 能够向池中添加进程而不等待其执行完毕就能再次执行循环
results.append(res)
print "--" * 10
pool.close() # 关闭pool, 则不会有新的进程添加进去
pool.join() # 必须在join之前close, 然后join等待pool中所有的线程执行完毕
print "All process done."
print "Return results: "
for i in results:
print i.get() # 获得进程的执行结果
结果：
"D:\Program Files\Anaconda2\python.exe" E:/pycharm/test/multiprocessing/v1.py -------------------- *msg: hello [0] *msg: hello [1] *msg: hello [2] *end *end *msg: hello [3] *msg: hello [4] *end *msg: hello [5] *end *end *msg: hello [6] *msg: hello [7] *end *msg: hello [8] *end *end *msg: hello [9] *end *end All process done. Return results: hello [0] return hello [1] return hello [2] return hello [3] return hello [4] return hello [5] return hello [6] return hello [7] return hello [8] return hello [9] return Process finished with exit code 0
map
函数原型：
map(func, iterable[, chunksize=None])
Pool类中的map方法，与内置的map函数用法和行为基本一致，它会使进程阻塞直到返回结果。
注意，虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。
# -*- coding:utf-8 -*-
import multiprocessing
import time
def func_with_return(msg):
print "*msg: ", msg
time.sleep(3)
print "*end"
return "{} return".format(msg)
if __name__ == "__main__":
# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
pool = multiprocessing.Pool(processes=3)
results = []
msgs = []
for i in range(10):
msg = "hello [{}]".format(i)
msgs.append(msg)
results = pool.map(func_with_return, msgs)
print "--" * 10
pool.close() # 关闭pool, 则不会有新的进程添加进去
pool.join() # 必须在join之前close, 然后join等待pool中所有的线程执行完毕
print "All process done."
print "Return results: "
for i in results:
print i # 获得进程的执行结果
执行结果：
"D:\Program Files\Anaconda2\python.exe" E:/pycharm/test/multiprocessing/v2.py *msg: hello [0] *msg: hello [1] *msg: hello [2] *end*end *msg: hello [3] *msg: hello [4] *end *msg: hello [5] *end*end *msg: hello [6] *msg: hello [7] *end *msg: hello [8] *end *end *msg: hello [9] *end *end -------------------- All process done. Return results: hello [0] return hello [1] return hello [2] return hello [3] return hello [4] return hello [5] return hello [6] return hello [7] return hello [8] return hello [9] return Process finished with exit code 0
注意：从执行结果中“—-”的位置可以看出，map之后，主进程是阻塞的，等待map的结果返回。
close()
关闭进程池（pool），使其不再接受新的任务。
terminate()
结束工作进程，不再处理未处理的任务。
join()
主进程阻塞等待子进程的退出，join方法必须在close或terminate之后使用。
进程间通信
多进程最复杂的部分在于进程间通信（IPC），其处理难度远高于线程通信，因此此处暂不展开，留作单独篇章记录。
利用multiprocessing实现一个最简单的分布式作业调度系统
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
总结
以上就是这篇文章的全部内容，希望本文的内容对大家的学习或者工作具有一定的参考学习价值。如果有疑问大家可以留言交流，谢谢大家对本站的支持。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
