未加星标

[Python]-14-多进程

字体大小 | |
[开发(python) 所属分类 开发(python) | 发布者 店小二05 | 时间 2016 | 作者 红领巾 ] 0人收藏点击收藏
引言

在硬件条件允许的情况下,使用多进程编程可以提高程序的执行效率,这篇文章将介绍如何让python程序实现多进程。

文章目录 0×1.Unix/linux系统与Fork() 0×2.使用multiprocessing模块实现多进程 a.使用Process类创建进程对象 c.使用Queue实现进程间通信 0×1.Unix/Linux系统与Fork()

注:这一部分所介绍的方法,仅适用于Unix/Linux操作系统,并不适用于windows系统,python在windows系统中的多进程实现可以使用跨平台的multiprocessing模块,请参考本文第二部分的内容;本文所有程序都是在Ubuntu系统环境下执行的,部分代码可能无法在Windows下运行。

说到多进程,就不得不介绍Unix/Linux系统中的一个经典系统调用fork(),当fork被程序调用时,会在程序调用的位置,记录下调用程序当前的所有状态信息(变量值等),再创建一个子进程,将当前程序(父进程)的状态信息传送给这个子进程,这个新建的子进程会复制父进程fork后面的所有程序过程,继续执行下面的内容,而此时,父进程本身也会继续执行fork下面的内容,原本一个进程执行的程序,在fork后变成了两个进程在执行。

Python的os模块封装中就包括了Unix/Linux系统的fork调用,下面用一个小程序来演示一下上面这些理论:

#!/usr/bin/env python3 #coding=utf-8 import os x=0 s="www.qingsword.com" rpid=os.fork() if rpid<0: print("fork调用失败。") elif rpid==0: print("我是子进程(%s),我的父进程是(%s)"%(os.getpid(),os.getppid())) x+=1 else: print("我是父进程(%s),我的子进程是(%s)"%(os.getpid(),rpid)) x+=1 print(s,x) #程序输出 我是父进程(7498),我的子进程是(7499) www.qingsword.com 1 我是子进程(7499),我的父进程是(7498) www.qingsword.com 1

从程序输出中不难发现,我们仅执行了一次程序,却同时运行了if判断的elif和else分之的内容,这就是fork的作用;

在rpid=os.fork()这条语句执行之前,程序只有一个进程在执行,此时程序已经初始化了两个变量,当执行fork语句后,程序创建了一个子进程,并且将这两个变量的状态传递给了这个子进程,子进程从if开始执行,直到执行完程序剩下的代码;

rpid=os.fork()之后,父进程也会继续执行,也就意味着,两个进程同时从if的位置开始执行剩下的代码,互不干扰;

os.getpid()能够获取当前进程的PID,os.getppid()能够获取当前进程的父进程PID;

那么,我们是如何知到rpid为0,就是子进程的呢?这是因为fork有三种不同的返回值:

1)在父进程中,fork返回新创建子进程的进程ID;

2)在子进程中,fork返回0;

3)如果出现错误,fork返回一个负值;

这样,我们就可以通过os.fork()这条语句执行后的返回值,来判断进程是父进程还是子进程,在子进程中,返回值是0,在父进程中,返回值是刚才新建的子进程的PID,如果创建进程失败,就会返回一个负值(通常如果系统限制了进程创建,或内存不足,或达到了系统进程上限,就会创建失败并返回一个负值);

在创建了新的进程后,这两个进程(父进程与子进程)执行没有固定的先后顺序,哪个进程先执行要看系统的进程调度策略;

其次,fork后,fork前已经初始化的那些变量会被复制成两份,两个进程中x的值都是0,并且两个进程的变量是独立的,存在不同的地址中,这就是为什么最后打印出来的x值是1的原因(每个进程的if流程都只执行了一次x+=1);

再来看一个实例:

#!/usr/bin/env python3 #coding=utf-8 import os x=1 while x<3: rpid=os.fork() if rpid==0: print("第%d次循环,我是子进程(%s),我的父进程是(%s),rpid值为(%s)"\ %(x,os.getpid(),os.getppid(),rpid)) else: print("第%d次循环,我是父进程(%s),我的子进程是(%s),rpid值为(%s)"\ %(x,os.getpid(),rpid,rpid)) x+=1 #程序输出 第1次循环,我是父进程(8790),我的子进程是(8791),rpid值为(8791) 第2次循环,我是父进程(8790),我的子进程是(8792),rpid值为(8792) 第1次循环,我是子进程(8791),我的父进程是(8790),rpid值为(0) 第2次循环,我是父进程(8791),我的子进程是(8793),rpid值为(8793) 第2次循环,我是子进程(8792),我的父进程是(8790),rpid值为(0) 第2次循环,我是子进程(8793),我的父进程是(8791),rpid值为(0)

第一次循环的时,当前进程(p8790)fork出了一个新的子进程(p8791),之后,两个进程分别先后进入了第二次循环,父进程(p8790)又fork出了一个新的子进程(p8792),而第一次循环中被创建的子进程(p8791)在第二次循环中也fork出了一个子进程(p8793),成了这个子进程的父进程。

大家可以尝试着增加循环次数,最后可以得到一个结论,循环n次所创建的子进程数量为:下标为k=1,上标为n的∑2^(k-1)

0×2.使用multiprocessing模块实现多进程

第一部分中所使用的fork仅仅是为了让大家了解什么是多进程,以及它的基本工作原理,这一部分将介绍一个跨平台的多进程模块"multiprocessing"。

a.使用Process类创建进程对象

Process是"multiprocessing"模块包含的一个类,可用于创建多进程程序,并可以指定创建的子进程对象去完成哪些代码块的执行。

Process语法结构如下:

Process([group [, target [, name [, args [, kwargs]]]]])

● target:表示这个进程实例所调用对象;

● args:表示调用对象的位置参数元组;

● kwargs:表示调用对象的关键字参数字典;

● name:为当前进程实例的别名;

● group:大多数情况下用不到;

Process类常用方法:

● is_alive():判断进程实例是否还在执行;

● join([timeout]):是否等待进程实例执行结束,或等待多少秒;

● start():启动进程实例(创建子进程);

● run():如果没有给定target参数,对这个对象调用start()方法时,就将执行对象中的run()方法;

● terminate():不管任务是否完成,立即终止;

Process类常用属性:

● name:当前进程实例别名,默认为Process-N,N为从1开始递增的整数;

● pid:当前进程实例的PID值;

● daemon:如果设置为True,父进程终止后子进程将自动终止,且自己不能产生新进程,必须在子进程实例start()之前设置;

下面用一个实例演示一下上面这些概念:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Process import time import os #两个子进程将会调用的两个方法 def worker_1(interval): print("worker_1,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start=time.time() time.sleep(interval) #程序将会被挂起interval秒 t_end=time.time() print("worker_1,执行时间为'%0.2f'秒"%(t_end-t_start)) def worker_2(interval): print("worker_2,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start=time.time() time.sleep(interval) t_end=time.time() print("worker_2,执行时间为'%0.2f'秒"%(t_end-t_start)) #输出当前程序的ID print("进程ID:%s"%os.getpid()) #创建两个进程对象,target指向这个进程对象要执行的对象名称,args后面的元组中,是要传递给worker_1方法的参数,因为worker_1方法就一个interval参数,这里传递一个整数2给它,如果不指定name参数,默认的进程对象名称为Process-N,N为一个递增的整数 p1=Process(target=worker_1,args=(2,)) p2=Process(target=worker_2,name="qing2",args=(1,)) #使用"进程对象名称.start()"来创建并执行一个子进程,这两个进程对象在start后,就会分别去执行worker_1和worker_2方法中的内容 p1.start() p2.start() #同时父进程仍然往下执行,如果p2进程还在执行,将会返回True print("p2.is_alive=%s"%p2.is_alive()) #输出p1和p2进程的别名和pid print("p1.name=%s"%p1.name) print("p1.pid=%s"%p1.pid) print("p2.name=%s"%p2.name) print("p2.pid=%s"%p2.pid) #join括号中不携带参数,表示父进程在这个位置要等待p1进程执行完成后,再继续执行下面的语句,一般用于进程间的数据同步,如果不写这一句,下面的is_alive判断将会是True,在WingIDE中也可能看不到worker_1执行完成后的print输出,因为主程序很可能先执行完成,但在shell(cmd)里面调用这个程序时可以完整的看到这个过程,大家可以尝试着将下面的这条语句改成p1.join(1),因为p2需要2秒以上才可能执行完成,父进程等待1秒很可能不能让p1完全执行完成,所以下面的print会输出True,即p1仍然在执行 p1.join() print("p1.is_alive=%s"%p1.is_alive()) #程序输出 进程ID:7609 p2.is_alive=True p1.name=Process-1 #p1使用了默认的别名 p1.pid=7610 p2.name=qing2 p2.pid=7611 worker_1,父进程(7609),当前进程(7610) worker_2,父进程(7609),当前进程(7611) worker_2,执行时间为'1.00'秒 #如果不使用join,在WingIDE中很可能看不到这一句输出,并且下面的is_alive会显示True worker_1,执行时间为'2.00'秒 p1.is_alive=False

再来看一个实例,下面这个程序,使用一个for循环获取当前程序正在运行的子进程列表:

#!/usr/bin/env python3 #coding=utf-8 import multiprocessing from multiprocessing import Process import time import os def worker_1(interval): print("worker_1,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start=time.time() time.sleep(interval) t_end=time.time() print("worker_1,执行时间为'%0.2f'秒"%(t_end-t_start)) print("进程ID:%s"%os.getpid()) #获取当前计算机CPU计算核心个数 print("当前计算机CPU计算核心个数:",multiprocessing.cpu_count()) #初始化4个进程对象,p1仅延迟1秒,前三个延迟3秒 p1=Process(target=worker_1,args=(3,)) p2=Process(target=worker_1,args=(3,)) p3=Process(target=worker_1,args=(3,)) p4=Process(target=worker_1,args=(1,)) p1.start() p2.start() p3.start() p4.start() p4.join() #等待p4执行完成后再往下执行 #因为p4已经执行完成,active_children()列表中只有前三个子进程 for ps in multiprocessing.active_children(): print(ps.name,ps.pid) p1.join() p2.join() p3.join() print("END") #程序输出 进程ID:8327 当前计算机CPU计算核心个数: 4 worker_1,父进程(8327),当前进程(8328) worker_1,父进程(8327),当前进程(8329) worker_1,父进程(8327),当前进程(8330) worker_1,父进程(8327),当前进程(8331) worker_1,执行时间为'1.00'秒 Process-3 8330 Process-1 8328 Process-2 8329 worker_1,执行时间为'3.00'秒 worker_1,执行时间为'3.00'秒 worker_1,执行时间为'3.00'秒 END

下面的代码演示了Process类的deamon属性的用法:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Process import time import os def worker_1(interval): print("worker_1,父进程(%s),当前进程(%s)"%(os.getppid(),os.getpid())) t_start=time.time() time.sleep(interval) t_end=time.time() print("worker_1,执行时间为'%0.2f'秒"%(t_end-t_start)) if __name__=="__main__": t_start=time.time() print("当前程序进程(%s)"%os.getpid()) p1=Process(target=worker_1,args=(2,)) p1.daemon=True #设置成True,只要父进程运行结束,子进程也会随即结束 p1.start() #p1.join() 添加这个方法就能让父进程等待子进程执行完成 t_stop=time.time() print("(%s)执行结束,耗时%0.2f"%(os.getpid(),t_stop-t_start)) #程序输出,从输出中可以看到,因为父进程短时间就执行结束了,所以子进程都还没来得及运行 当前程序进程(13921) (13921)执行结束,耗时0.01

如果想要设置daemon参数为True,又想在父进程结束前执行完子进程,可以在主进程中加入join方法,等待子进程执行完成。

现在我们已经可以任意的创建子进程去执行指定目标代码块了,实际上,创建新的进程还能够使用类的方式,可以自定义一个类,继承Process类,每次实例化这个类的时候,就等同于实例化一个进程对象,请看下面的实例:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Process import time import os #继承Process类 class Process_Class(Process): #因为Process类本身也有__init__方法,这个子类相当于重写了这个方法,但这样就会带来一个问题,我们并没有完全的初始化一个Process类,所以就不能使用从这个类继承的一些方法和属性,最好的方法就是将继承类本身传递给Process.__init__方法,完成这些初始化操作 def __init__(self,interval): Process.__init__(self) self.interval=interval #重写了Process类的run()方法 def run(self): print("子进程(%s) 开始执行,父进程为(%s)"%(os.getpid(),os.getppid())) t_start=time.time() time.sleep(self.interval) t_stop=time.time() print("(%s)执行结束,耗时%0.2f秒"%(os.getpid(),t_stop-t_start)) if __name__=="__main__": t_start=time.time() print("当前程序进程(%s)"%os.getpid()) p1=Process_Class(2) #对一个不包含target属性的Process类执行start()方法,就会运行这个类中的run()方法,所以这里会执行p1.run() p1.start() p1.join() t_stop=time.time() print("(%s)执行结束,耗时%0.2f"%(os.getpid(),t_stop-t_start)) #程序输出 当前程序进程(12153) 子进程(12154) 开始执行,父进程为(12153) (12154)执行结束,耗时2.00秒 (12153)执行结束,耗时2.01 b.使用进程池Pool

当需要创建的子进程数量不多时,可以直接利用multiprocessing中的Process动态成生多个进程,但如果是上百甚至上千个目标,手动的去创建进程的工作量巨大,此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时,可以指定一个最大进程数,当有新的请求提交到Pool中时,如果池还没有满,那么就会创建一个新的进程用来执行该请求;但如果池中的进程数已经达到指定的最大值,那么该请求就会等待,直到池中有进程结束,才会创建新的进程来执行,请看下面的实例:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Pool import os,time,random def worker(msg): t_start=time.time() print(msg,"开始执行") #random.random()随机生成0~1之间的浮点数 time.sleep(random.random()*2) t_stop=time.time() print(msg,"执行完毕,耗时%0.2f"%(t_stop-t_start)) po=Pool(3) #定义一个进程池,最大进程数3 for i in range(0,5): #Pool.apply_async(要调用的目标,(传递给目标的参数元祖,)),每次循环将会创建一个新的子进程去调用目标,直到达到进程池最大进程数的限制 po.apply_async(worker,(i,)) print("www.qingsword.com") po.close() #关闭进程池,关闭后po不再接收新的请求 po.join() #等待po中所有子进程执行完成,必须放在close语句之后 print("End") #程序输出 www.qingsword.com 0 开始执行 1 开始执行 2 开始执行 #因为Pool(3)所以并行运行了三个子进程 2 执行完毕,耗时0.96 3 开始执行 #直到上面三个子进程中的一个运行完成,再创建新的子进程 3 执行完毕,耗时0.08 4 开始执行 0 执行完毕,耗时1.61 4 执行完毕,耗时0.67 1 执行完毕,耗时1.86 End

multiprocessing.Pool常用函数解析:

● apply_async(func[, args[, kwds[, callback]]]) :使用非阻塞方式调用func(并行执行,堵塞方式必须等待上一个进程退出才能执行下一个进程),args为传递给func的参数列表,kwds为传递给func的关键字参数列表; ● apply(func[, args[, kwds]]):使用阻塞方式调用func(稍后会用实例演示,看输出就能明白两者区别);

● close():关闭Pool,使其不再接受新的任务;

● terminate():不管任务是否完成,立即终止;

● join():主进程阻塞,等待子进程的退出, 必须在close或terminate之后使用;

下面来看一个阻塞方式调用的实例:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Pool import os,time,random def worker(msg): t_start=time.time() print(msg,"开始执行") time.sleep(random.random()*2) t_stop=time.time() print(msg,"执行完毕,耗时%0.2f"%(t_stop-t_start)) po=Pool(3) for i in range(0,5): po.apply(worker,(i,)) #唯一的区别在这里 print("www.qingsword.com") po.close() po.join() print("End") #程序输出,从输出中可以看出,每个子进程执行完成后再创建下一个子进程,父进程也被阻塞,必须等待所有子进程调用完成后,再继续执行下面的内容 0 开始执行 0 执行完毕,耗时0.27 1 开始执行 1 执行完毕,耗时0.06 2 开始执行 2 执行完毕,耗时1.19 3 开始执行 3 执行完毕,耗时1.36 4 开始执行 4 执行完毕,耗时0.70 www.qingsword.com End c.使用Queue实现进程间通信

可以使用multiprocessing模块的Queue实现多进程之间的数据传递,Queue本身是一个消息列队程序,首先用一个小实例来演示一下Queue的工作原理:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Queue q=Queue(3) #初始化一个Queue对象,最多可接收三条put消息 q.put("消息1") q.put("消息2") print(q.full()) #False q.put("消息3") print(q.full()) #True #因为消息列队已满下面的try都会抛出异常,第一个try会等待2秒后再抛出异常,第二个Try会立刻抛出异常 try: q.put("消息4",True,2) except: print("消息列队已满,现有消息数量:%s"%q.qsize()) try: q.put_nowait("消息4") except: print("消息列队已满,现有消息数量:%s"%q.qsize()) #推荐的方式,先判断消息列队是否已满,再写入 if not q.full(): q.put_nowait("消息4") #读取消息时,先判断消息列队是否为空,再读取 if not q.empty(): for i in range(q.qsize()): print(q.get_nowait()) #程序输出 False True 消息列队已满,现有消息数量:3 消息列队已满,现有消息数量:3 消息1 消息2 消息3

下面是Queue对象中所包含的常用方法的功能简介:

初始化Queue()对象时(例如:q=Queue()),若括号中没有指定最大可接收的消息数量,或数量为负值,那么就代表可接受的消息数量没有上限(直到内存的尽头);

● Queue.qsize():返回当前队列包含的消息数量;

● Queue.empty():如果队列为空,返回True,反之False ;

● Queue.full():如果队列满了,返回True,反之False;

● Queue.get([block[, timeout]]):获取队列中的一条消息,然后将其从列队中移除,block默认值为True;

1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果为空,此时程序将被阻塞(停在读取状态),直到从消息列队读到消息为止,如果设置了timeout,则会等待timeout秒,若还没读取到任何消息,则抛出"Queue.Empty"异常;

2)如果block值为False,消息列队如果为空,则会立刻抛出"Queue.Empty"异常;

● Queue.get_nowait():相当Queue.get(False);

● Queue.put(item,[block[, timeout]]):将item消息写入队列,block默认值为True;

1)如果block使用默认值,且没有设置timeout(单位秒),消息列队如果已经没有空间可写入,此时程序将被阻塞(停在写入状态),直到从消息列队腾出空间为止,如果设置了timeout,则会等待timeout秒,若还没空间,则抛出"Queue.Full"异常;

2)如果block值为False,消息列队如果没有空间可写入,则会立刻抛出"Queue.Full"异常;

● Queue.put_nowait(item):相当Queue.put(item, False);

在了解了这些基本原理后,来看看Queue如何实现不同进程间数据的传递:

#!/usr/bin/env python3 #coding=utf-8 from multiprocessing import Queue,Process import os,time,random #用于读取列队中的消息打印到屏幕 def reader(q): print("reader启动(%s),父进程为(%s)"%(os.getpid(),os.getppid())) #一个死循环,结束时需要使用terminate()方法强行结束,q.get(True)中设置了block值为True,并且没有指定等待时间 while True: print("reader从Queue获取到消息:%s"%q.get(True)) #用于向列队中put消息 def writer(q): print("writer启动(%s),父进程为(%s)"%(os.getpid(),os.getppid())) L=(x for x in "qing") for i in L: q.put(i) time.sleep(random.random()*3) #设置一个延迟来模拟进程put消息的时间差 if __name__=="__main__": print("(%s) start"%os.getpid()) q=Queue() #初始化一个Queue对象没有指定消息列队上限 #将q对象作为参数传递给下面两个进程 pr=Process(target=reader,args=(q,)) pw=Process(target=writer,args=(q,)) pr.start() pw.start() pw.join() pr.terminate() #因为pr不能自己终止,所以只能强行终止 print("(%s) End"%os.getpid()) #程序输出 (11592) start reader启动(11593),父进程为(11592) writer启动(11594),父进程为(11592) reader从Queue获取到消息:q reader从Queue获取到消息:i reader从Queue获取到消息:n reader从Queue获取到消息:g (11592) End

上面这种方法只适用于普通的Process进程创建,如果要使用Pool创建进程,就需要使用multiprocessing.Manager()中的Queue(),而不是multiprocessing.Queue(),否则会得到一条如下的错误信息:

RuntimeError: Queue objects should only be shared between processes through inheritance.

下面的实例演示了进程池中的进程如何通信:

#!/usr/bin/env python3 #coding=utf-8 #修改import中的Queue为Manager from multiprocessing import Manager,Pool import os,time,random def reader(q): print("reader启动(%s),父进程为(%s)"%(os.getpid(),os.getppid())) for i in range(q.qsize()): print("reader从Queue获取到消息:%s"%q.get(True)) def writer(q): print("writer启动(%s),父进程为(%s)"%(os.getpid(),os.getppid())) L=(x for x in "qing") for i in L: q.put(i) if __name__=="__main__": print("(%s) start"%os.getpid()) q=Manager().Queue() #使用Manager中的Queue来初始化 po=Pool() #使用阻塞模式创建进程,这样就不需要在reader中使用死循环了,可以让writer完全执行完成后,再用reader去读取 po.apply(writer,(q,)) po.apply(reader,(q,)) po.close() po.join() print("(%s) End"%os.getpid()) #程序输出 (11971) start writer启动(11978),父进程为(11971) reader启动(11977),父进程为(11971) reader从Queue获取到消息:q reader从Queue获取到消息:i reader从Queue获取到消息:n reader从Queue获取到消息:g (11971) End

本文开发(python)相关术语:python基础教程 python多线程 web开发工程师 软件开发工程师 软件开发流程

分页:12
转载请注明
本文标题:[Python]-14-多进程
本站链接:http://www.codesec.net/view/483109.html
分享请点击:


1.凡CodeSecTeam转载的文章,均出自其它媒体或其他官网介绍,目的在于传递更多的信息,并不代表本站赞同其观点和其真实性负责;
2.转载的文章仅代表原创作者观点,与本站无关。其原创性以及文中陈述文字和内容未经本站证实,本站对该文以及其中全部或者部分内容、文字的真实性、完整性、及时性,不作出任何保证或承若;
3.如本站转载稿涉及版权等问题,请作者及时联系本站,我们会及时处理。
登录后可拥有收藏文章、关注作者等权限...
技术大类 技术大类 | 开发(python) | 评论(0) | 阅读(43)