Python multiprocessing多进程原理与应用示例
本文实例讲述了Pythonmultiprocessing多进程原理与应用。分享给大家供大家参考,具体如下:
multiprocessing包是Python中的多进程管理包,可以利用multiprocessing.Process对象来创建进程,Process对象拥有is_alive()、join([timeout])、run()、start()、terminate()等方法。
multprocessing模块的核心就是使管理进程像管理线程一样方便,每个进程有自己独立的GIL,所以不存在进程间争抢GIL的问题,在多核CPU环境中,可以大大提高运行效率。
multiprocessing使用示例:
importmultiprocessing importtime importcv2 defdaemon1(image): name=multiprocessing.current_process().name foriinrange(50): image=cv2.GaussianBlur(image,(3,3),1) time.sleep(0.1) print'daemon1done!' cv2.imshow('daemon1',image) defdaemon2(image): name=multiprocessing.current_process().name foriinrange(50): image=cv2.GaussianBlur(image,(3,3),1) time.sleep(0.5) print'daemon2done!' cv2.imshow('daemon2',image) if__name__=='__main__': t1=time.time() number_kernel=multiprocessing.cpu_count() print'Wehave{0}kernels'.format(number_kernel) p1=multiprocessing.Process(name='daemon1', target=daemon1,args=(cv2.imread('./p1.jpg'),)) p1.daemon=False p2=multiprocessing.Process(name='daemon2', target=daemon2,args=(cv2.imread('./p2.jpg'),)) p2.daemon=False p1.start() p2.start() print'p1is{0}'.format(p1.is_alive()) p1.terminate() p1.join() print'p1is{0}'.format(p1.is_alive()) print'p2is{0}'.format(p2.is_alive()) p2.join() t2=time.time() print'!!!!!!!!!!!!!!!!!!!!OK!!!!!!!!!!!!!!!!!!!!!' print'totaltimeis{0}'.format(t2-t1) print'p1.exitcode={0}'.format(p1.exitcode) print'p2.exitcode={0}'.format(p2.exitcode)
multiprocessing中Process是一个类,用于创建进程,以及定义进程的方法,Process类的构造函数是:
def__init__(self,group=None,target=None,name=None,args=(),kwargs={})
参数含义:
- group:进程所属组,基本不用
- target:创建进程关联的对象,需要传入要多进程处理的函数名
- name:定义进程的名称
- args:表示与target相关联的函数的传入参数,可以传入多个,注意args是一个元组,如果传入的参数只有一个,需要表示为args=(element1,)
- kwargs:表示调用对象的字典
程序解读:
- multiprocessing.cpu_count():返回机器上cpu核的总数量
- p1.daemon=False:定义子进程的运行属性,如果.daemon设置为False表示子进程可以在主进程完成之后继续执行;如果.daemon设置为True,表示子进程随着主进程的结束而结束;必须在start之前设置;
- p1.start():开始执行子进程p1
- p1.join():定义子进程p1的运行阻塞主进程,只有p1子进程执行完成之后才会继续执行join之后的主进程,但是子进程间互相不受join影响。
- 可以定义子进程阻塞主进程的时间--p1.join(100),超时之后,主进程不再等待,开始执行。join()需要放在start()方法之后;
- p1.terminate():终止子进程的执行,其后要跟上jion()方法更新子进程的状态;
- p1.exitcode:进程的退出状态: ==0未生成任何错误,正常退出; >0进程有一个错误,并以该错误码退出; <0进程由一个-1*exitcode信号结束
在multiprocessing中使用pool
如果需要多个子进程时,使用进程池(pool)来(自动)管理各个子进程更加方便:
frommultiprocessingimportPool importos,time deflong_time_task(name): print'Runtask{0}({1})'.format(name,os.getpid()) start=time.time() time.sleep(3) end=time.time() print'Task{0}runs{1:.2f}seconds.'.format(name,end-start) if__name__=='__main__': print'Parentprocess({0})'.format(os.getpid) p=Pool() foriinrange(12): p.apply_async(long_time_task,args=(i,)) print'Waitingforallsubprocessesdone...' p.close() p.join() print'Allsubprocessesdone.'
与Process类创建进程的方法不同,Pool是通过apply_async(func,args=(args))方法创建进程,一个进程池中能同时运行的任务数是机器上CPU核的总数量n_kernel,如果创建的子进程数大于n_kernel,则同时执行n_kernel个进程,这n_kernel中某个进程完成之后才会启动下一个进程。
- os.getpid()是获取当前执行的进程的ID
- p.close()方法是关掉进程池,表示不能再继续向进程池添加进程了。
- p.join()方法是子进程阻塞主进程,必须在调用p.close()关闭进程池之后才能调用join()方法
多个子进程间的通信
多个子进程间的通信要用到multiprocessing.Queue,Queue的特性是它是一个消息队列。比如有以下的需求,一个子进程向队列中写数据,另外一个进程从队列中取数据的例子:
frommultiprocessingimportProcess,Queue importos,time,random defwrite(q): forvaluein['A','B','C']: print'Put{0}toqueue...'.format(value) q.put(value) time.sleep(random.random()) defread(q): whileTrue: ifnotq.empty(): value=q.get(True) print'Get{0}fromqueue.'.format(value) time.sleep(random.random()) else: break if__name__=='__main__': q=multiprocessing.Queue() pw=Process(target=write,args=(q,)) pr=Process(target=read,args=(q,)) pw.start() pw.join() pr.start() pr.join()
Queue使用方法:
- Queue.qsize():返回当前队列包含的消息数量;
- Queue.empty():如果队列为空,返回True,反之False;
- Queue.full():如果队列满了,返回True,反之False;
- Queue.get():获取队列中的一条消息,然后将其从列队中移除,可传参超时时长;
- Queue.get_nowait():相当Queue.get(False),取不到值时触发异常:Empty;
- Queue.put():将一个值添加进数列,可传参超时时长;
- Queue.put_nowait():相当于Queue.get(False),当队列满了时报错:Full;
在进程池Pool中,使用Queue会出错,需要使用Manager.Queue:
frommultiprocessingimportProcess,Queue importos,time,random defwrite(q): forvaluein['A','B','C']: print'Put{0}toqueue...'.format(value) q.put(value) time.sleep(random.random()) defread(q): whileTrue: ifnotq.empty(): value=q.get(True) print'Get{0}fromqueue.'.format(value) time.sleep(random.random()) else: break if__name__=='__main__': manager=multiprocessing.Manager() q=manager.Queue() p=Pool() pw=p.apply_async(write,args=(q,)) time.sleep(2) pr=p.apply_async(read,args=(q,)) p.close() p.join() ifnotq.empty(): print'qisnotempty...' else: print'qisempty...' print'OK' ifnotq.empty(): print'qisnotempty...' else: print'qisempty...' print'done...'
父进程与子进程共享内存
定义普通的变量,不能实现在父进程和子进程之间共享:
importmultiprocessing frommultiprocessingimportPool defchangevalue(n,a): n=3.14 a[0]=5 if__name__=='__main__': num=0 arr=range(10) p=Pool() p1=p.apply_async(changevalue,args=(num,arr)) p.close() p.join() printnum printarr[:]
结果输出num的值还是在父进程中定义的0,arr的第一个元素值还是0。
使用multiprocessing创建共享对象:
importmultiprocessing defchangevalue(n,a): n.value=3.14 a[0]=5 if__name__=='__main__': num=multiprocessing.Value('d',0.0) arr=multiprocessing.Array('i',range(10)) p=multiprocessing.Process(target=changevalue,args=(num,arr)) p.start() p.join() printnum.value printarr[:]
结果输出num的值是在子进程中修改的3.14,arr的第一个元素值更改为5。
共享内存在Pool中的使用:
importmultiprocessing frommultiprocessingimportPool defchangevalue(n,a): n.value=3.14 a[0]=5 if__name__=='__main__': num=multiprocessing.Value('d',0.0) arr=multiprocessing.Array('i',range(10)) p=Pool() p1=p.apply_async(changevalue,args=(num,arr)) p.close() p.join() printnum.value printarr[:]
更多关于Python相关内容感兴趣的读者可查看本站专题:《Python进程与线程操作技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》、《Python+MySQL数据库程序设计入门教程》及《Python常见数据库操作技巧汇总》
希望本文所述对大家Python程序设计有所帮助。