Python利用multiprocessing实现最简单的分布式作业调度系统实例
介绍
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的很大一部份与threading使用同一套API，只不过换到了多进程的情境。
这个模块表示像线程一样管理进程，这个是multiprocessing的核心，它与threading很相似，对多核CPU的利用率会比threading好的多。
看一下Process类的构造方法：
__init__(self,group=None,target=None,name=None,args=(),kwargs={})
参数说明：
- group：进程所属组。基本不用
- target：表示调用对象。
- args：表示调用对象的位置参数元组。
- name：别名
- kwargs：表示调用对象的字典。
创建进程的简单实例：
#coding=utf-8 importmultiprocessing defdo(n): #获取当前线程的名字 name=multiprocessing.current_process().name printname,'starting' print"worker",n return if__name__=='__main__': numList=[] foriinxrange(5): p=multiprocessing.Process(target=do,args=(i,)) numList.append(p) p.start() p.join() print"Processend."
执行结果：
Process-1starting worker0 Processend. Process-2starting worker1 Processend. Process-3starting worker2 Processend. Process-4starting worker3 Processend. Process-5starting worker4 Processend.
创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，并用其start()方法启动，join()方法表示等待子进程结束以后再继续往下运行，通常用于进程间的同步。
注意：
在Windows上要想使用进程模块，就必须把有关进程的代码写在当前.py文件的if__name__==‘__main__':语句的下面，才能正常使用Windows下的进程模块。Unix/Linux下则不需要。
multiprocess.Pool
当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个目标，手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。
Pool可以提供指定数量的进程供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。
apply_async和apply
函数原型：
apply_async(func[,args=()[,kwds={}[,callback=None]]])
二者都是向进程池中添加新的进程，不同的时，apply每次添加新的进程时，主进程和新的进程会并行执行，但是主进程会阻塞，直到新进程的函数执行结束。这是很低效的，所以python3.x之后不再使用
apply_async和apply功能相同，但是主进程不会阻塞。
#-*-coding:utf-8-*-
importmultiprocessing
importtime
deffunc(msg):
print"*msg:",msg
time.sleep(3)
print"*end"
if__name__=="__main__":
#维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
pool=multiprocessing.Pool(processes=3)
foriinrange(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"Allprocessdone."
运行结果：
"D:\ProgramFiles\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 Allprocessdone. Processfinishedwithexitcode0
获得进程的执行结果
#-*-coding:utf-8-*-
importmultiprocessing
importtime
deffunc_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=[]
foriinrange(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"Allprocessdone."
print"Returnresults:"
foriinresults:
printi.get()#获得进程的执行结果
结果：
"D:\ProgramFiles\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 Allprocessdone. Returnresults: 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 Processfinishedwithexitcode0
map
函数原型：
map(func,iterable[,chunksize=None])
Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程阻塞直到返回结果。
注意，虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。
#-*-coding:utf-8-*-
importmultiprocessing
importtime
deffunc_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=[]
foriinrange(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"Allprocessdone."
print"Returnresults:"
foriinresults:
printi#获得进程的执行结果
执行结果：
"D:\ProgramFiles\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 -------------------- Allprocessdone. Returnresults: 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 Processfinishedwithexitcode0
注意：执行结果中“—-”的位置，可以看到，map之后，主进程是阻塞的，等待map的结果返回
close()
关闭进程池（pool），使其不在接受新的任务。
terminate()
结束工作进程，不在处理未处理的任务。
join()
主进程阻塞等待子进程的退出，join方法必须在close或terminate之后使用。
进程间通信
多进程最麻烦的地方就是进程间通信，IPC比线程通信要难处理的多，所以留作单独一篇来记录
利用multiprocessing实现一个最简单的分布式作业调度系统
Job
首先创建一个Job类，为了测试简单，只包含一个jobid属性，将来可以封装一些作业状态，作业命令，执行用户等属性。
job.py
#!/usr/bin/envpython #-*-coding:utf-8-*- classJob: def__init__(self,job_id): self.job_id=job_id
Master
Master用来派发作业和显示运行完成的作业信息
master.py
#!/usr/bin/envpython
#-*-coding:utf-8-*-
fromQueueimportQueue
frommultiprocessing.managersimportBaseManager
fromjobimportJob
classMaster:
def__init__(self):
#派发出去的作业队列
self.dispatched_job_queue=Queue()
#完成的作业队列
self.finished_job_queue=Queue()
defget_dispatched_job_queue(self):
returnself.dispatched_job_queue
defget_finished_job_queue(self):
returnself.finished_job_queue
defstart(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
whileTrue:
foriinrange(0,10):
job_id=job_id+1
job=Job(job_id)
print('Dispatchjob:%s'%job.job_id)
dispatched_jobs.put(job)
whilenotdispatched_jobs.empty():
job=finished_jobs.get(60)
print('FinishedJob:%s'%job.job_id)
manager.shutdown()
if__name__=="__main__":
master=Master()
master.start()
Slave
Slave用来运行master派发的作业并将结果返回
slave.py
#!/usr/bin/envpython
#-*-coding:utf-8-*-
importtime
fromQueueimportQueue
frommultiprocessing.managersimportBaseManager
fromjobimportJob
classSlave:
def__init__(self):
#派发出去的作业队列
self.dispatched_job_queue=Queue()
#完成的作业队列
self.finished_job_queue=Queue()
defstart(self):
#把派发作业队列和完成作业队列注册到网络上
BaseManager.register('get_dispatched_job_queue')
BaseManager.register('get_finished_job_queue')
#连接master
server='127.0.0.1'
print('Connecttoserver%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()
#运行作业并返回结果，这里只是模拟作业运行，所以返回的是接收到的作业
whileTrue:
job=dispatched_jobs.get(timeout=1)
print('Runjob:%s'%job.job_id)
time.sleep(1)
finished_jobs.put(job)
if__name__=="__main__":
slave=Slave()
slave.start()
测试
分别打开三个linux终端，第一个终端运行master，第二个和第三个终端用了运行slave，运行结果如下
master
$pythonmaster.py Dispatchjob:1 Dispatchjob:2 Dispatchjob:3 Dispatchjob:4 Dispatchjob:5 Dispatchjob:6 Dispatchjob:7 Dispatchjob:8 Dispatchjob:9 Dispatchjob:10 FinishedJob:1 FinishedJob:2 FinishedJob:3 FinishedJob:4 FinishedJob:5 FinishedJob:6 FinishedJob:7 FinishedJob:8 FinishedJob:9 Dispatchjob:11 Dispatchjob:12 Dispatchjob:13 Dispatchjob:14 Dispatchjob:15 Dispatchjob:16 Dispatchjob:17 Dispatchjob:18 Dispatchjob:19 Dispatchjob:20 FinishedJob:10 FinishedJob:11 FinishedJob:12 FinishedJob:13 FinishedJob:14 FinishedJob:15 FinishedJob:16 FinishedJob:17 FinishedJob:18 Dispatchjob:21 Dispatchjob:22 Dispatchjob:23 Dispatchjob:24 Dispatchjob:25 Dispatchjob:26 Dispatchjob:27 Dispatchjob:28 Dispatchjob:29 Dispatchjob:30
slave1
$pythonslave.py Connecttoserver127.0.0.1... Runjob:1 Runjob:2 Runjob:3 Runjob:5 Runjob:7 Runjob:9 Runjob:11 Runjob:13 Runjob:15 Runjob:17 Runjob:19 Runjob:21 Runjob:23
slave2
$pythonslave.py Connecttoserver127.0.0.1... Runjob:4 Runjob:6 Runjob:8 Runjob:10 Runjob:12 Runjob:14 Runjob:16 Runjob:18 Runjob:20 Runjob:22 Runjob:24
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对毛票票的支持。