时间:2021-05-22
Python中的多线程其实并不是真正的多线程,如果想要充分地使用多核CPU的资源,在python中大部分情况需要使用多进程。Python提供了非常好用的多进程包multiprocessing,只需要定义一个函数,Python会完成其他所有事情。借助这个包,可以轻松完成从单进程到并发执行的转换。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步,提供了Process、Queue、Pipe、Lock等组件。
引例:
如之前创建多进程的例子
# -*- coding:utf-8 -*-from multiprocessing import Process,Poolimport os,timedef run_proc(name): ##定义一个函数用于进程调用 for i in range(5): time.sleep(0.2) #休眠0.2秒 print 'Run child process %s (%s)' % (name, os.getpid())#执行一次该函数共需1秒的时间if __name__ =='__main__': #执行主进程 print 'Run the main process (%s).' % (os.getpid()) mainStart = time.time() #记录主进程开始的时间 p = Pool(8) #开辟进程池 for i in range(16): #开辟14个进程 p.apply_async(run_proc,args=('Process'+str(i),))#每个进程都调用run_proc函数, #args表示给该函数传递的参数。 print 'Waiting for all subprocesses done ...' p.close() #关闭进程池 p.join() #等待开辟的所有进程执行完后,主进程才继续往下执行 print 'All subprocesses done' mainEnd = time.time() #记录主进程结束时间 print 'All process ran %0.2f seconds.' % (mainEnd-mainStart) #主进程执行时间运行结果:
Run the main process (36652). Waiting for all subprocesses done … Run child process Process0 (36708)Run child process Process1 (36748)Run child process Process3 (36736) Run child process Process2 (36716) Run child process Process4 (36768)如第3行的输出,偶尔会出现这样不如意的输入格式,为什么呢?
原因是多个进程争用打印输出资源的结果。前一个进程为来得急输出换行符,该资源就切换给了另一个进程使用,致使两个进程输出在同一行上,而前一个进程的换行符在下一次获得资源时才打印输出。
Lock
为了避免这种情况,需在进程进入临界区(使进程进入临界资源的那段代码,称为临界区)时加锁。
可以向如下这样添加锁后看看执行效果:
Semaphore
Semaphore为信号量机制。当共享的资源拥有多个时,可用Semaphore来实现进程同步。其用法和Lock差不多,s = Semaphore(N),每执行一次s.acquire(),该资源的可用个数将减少1,当资源个数已为0时,就进入阻塞;每执行一次s.release(),占用的资源被释放,该资源的可用个数增加1。
多进程的通信(信息交互)
不同进程之间进行数据交互,可能不少刚开始接触多进程的同学会想到共享全局变量的方式,这样通过向全局变量写入和读取信息便能实现信息交互。但是很遗憾,并不能这样实现。
下面通过例子,加深对那篇文章的理解:
# -*- coding:utf-8 -*-from multiprocessing import Process, Poolimport osimport timeL1 = [1, 2, 3]def add(a, b): global L1 L1 += range(a, b) print L1if __name__ == '__main__': p1 = Process(target=add, args=(20, 30)) p2 = Process(target=add, args=(30, 40)) p1.start() p2.start() p1.join() p2.join() print L1输出结果:
[1, 2, 3, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
[1, 2, 3, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
[1, 2, 3]
该程序的原本目的是想将两个子进程生成的列表加到全局变量L1中,但用该方法并不能达到想要的效果。既然不能通过全局变量来实现不同进程间的信息交互,那有什么办法呢。
mutiprocessing为我们可以通过Queue和Pipe来实现进程间的通信。
Queue
按上面的例子通过Queue来实现:
# -*- coding:utf-8 -*-from multiprocessing import Process, Queue, LockL = [1, 2, 3]def add(q, lock, a, b): lock.acquire() # 加锁避免写入时出现不可预知的错误 L1 = range(a, b) lock.release() q.put(L1) print L1if __name__ == '__main__': q = Queue() lock = Lock() p1 = Process(target=add, args=(q, lock, 20, 30)) p2 = Process(target=add, args=(q, lock, 30, 40)) p1.start() p2.start() p1.join() p2.join() L += q.get() + q.get() print L执行结果:
[20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
[30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
[1, 2, 3, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
下面介绍Queue的常用方法:
Pipe
Pipe管道,可以是单向(half-duplex),也可以是双向(duplex)。我们通过mutiprocessing.Pipe(duplex=False)创建单向管道 (默认为双向)。双向Pipe允许两端的进即可以发送又可以接受;单向的Pipe只允许前面的端口用于接收,后面的端口用于发送。
下面给出例子:
# -*- coding:utf-8 -*-from multiprocessing import Process, Pipedef proc1(pipe): s = 'Hello,This is proc1' pipe.send(s)def proc2(pipe): while True: print "proc2 recieve:", pipe.recv()if __name__ == "__main__": pipe = Pipe() p1 = Process(target=proc1, args=(pipe[0],)) p2 = Process(target=proc2, args=(pipe[1],)) p1.start() p2.start() p1.join() p2.join(2) #限制执行时间最多为2秒 print '\nend all processes.'执行结果如下:
proc2 recieve: Hello,This is proc1
proc2 recieve:
end all processes.
当第二行输出后,因为管道中没有数据传来,Proc2处于阻塞状态,2秒后被强制结束。
以下是单向管道的例子,注意pipe[0],pipe[1]的分配。
# -*- coding:utf-8 -*-from multiprocessing import Process, Pipedef proc1(pipe): s = 'Hello,This is proc1' pipe.send(s)def proc2(pipe): while True: print "proc2 recieve:", pipe.recv()if __name__ == "__main__": pipe = Pipe(duplex=False) p1 = Process(target=proc1, args=(pipe[1],)) #pipe[1]为发送端 p2 = Process(target=proc2, args=(pipe[0],)) #pipe[0]为接收端 p1.start() p2.start() p1.join() p2.join(2) # 限制执行时间最多为2秒 print '\nend all processes.'执行结果同上。
强大的Manage
Queue和Pipe实现的数据共享方式只支持两种结构 Value 和 Array。Python中提供了强大的Manage专门用来做数据共享,其支持的类型非常多,包括: Value,Array,list, dict,Queue, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event等
其用法如下:
from multiprocessing import Process, Managerdef func(dt, lt): for i in range(10): key = 'arg' + str(i) dt[key] = i * i lt += range(11, 16)if __name__ == "__main__": manager = Manager() dt = manager.dict() lt = manager.list() p = Process(target=func, args=(dt, lt)) p.start() p.join() print dt, '\n', lt执行结果:
{‘arg8': 64, ‘arg9': 81, ‘arg0': 0, ‘arg1': 1, ‘arg2': 4, ‘arg3': 9, ‘arg4': 16, ‘arg5': 25, ‘arg6': 36, ‘arg7': 49}
[11, 12, 13, 14, 15]
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
进程同步用来实现程序并发执行时候的可再现性。一.进程同步及异步的概念1.进程同步:就是在发出一个功能调用时,在没有得到结果之前,该调用就不返回。也就是必须一件一
本文实例讲述了python执行子进程实现进程间通信的方法。分享给大家供大家参考。具体实现方法如下:a.py:importsubprocess,timesubpr
目标:优化代码,利用多进程,进行近实时预处理、网络预测及后处理:本人尝试了pytorch的multiprocessing,进行多进程同步处理以上任务。fromt
前言单JVM内同步好办,直接用JDK提供的锁就可以了,但是跨进程同步靠这个肯定是不可能的,这种情况下肯定要借助第三方,我这里实现用Redis,当然还有很多其他的
先看看下面的几个方法:star()方法启动进程,join()方法实现进程间的同步,等待所有进程退出。close()用来阻止多余的进程涌入进程池Pool造成进程阻