本文实例讲述了Python multiprocessing多进程原理与应用。分享给大家供大家参考，具体如下：
multiprocessing包是Python中的多进程管理包,可以利用multiprocessing.Process对象来创建进程,Process对象拥有is_alive()、join([timeout])、run()、start()、terminate()等方法。
multprocessing模块的核心就是使管理进程像管理线程一样方便，每个进程有自己独立的GIL，所以不存在进程间争抢GIL的问题，在多核CPU环境中，可以大大提高运行效率。
multiprocessing使用示例：
import multiprocessingimport timeimport cv2def daemon1(image): name = multiprocessing.current_process().name for i in range(50): image = cv2.GaussianBlur(image, (3, 3), 1) time.sleep(0.1) print 'daemon1 done!' cv2.imshow('daemon1', image)def daemon2(image): name = multiprocessing.current_process().name for i in range(50): image = cv2.GaussianBlur(image, (3, 3), 1) time.sleep(0.5) print 'daemon2 done!' cv2.imshow('daemon2', image)if __name__ == '__main__': t1 = time.time() number_kernel = multiprocessing.cpu_count() print 'We have {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 'p1 is {0}'.format(p1.is_alive()) p1.terminate() p1.join() print 'p1 is {0}'.format(p1.is_alive()) print 'p2 is {0}'.format(p2.is_alive()) p2.join() t2 = time.time() print '!!!!!!!!!!!!!!!!!!!!OK!!!!!!!!!!!!!!!!!!!!!' print 'total time is {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)来（自动）管理各个子进程更加方便：
from multiprocessing import Poolimport os, timedef long_time_task(name): print 'Run task {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 'Parent process ({0})'.format(os.getpid) p = Pool() for i in range(12): p.apply_async(long_time_task, args=(i,)) print 'Waiting for all subprocesses done...' p.close() p.join() print 'All subprocesses done.' 与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的特性是它是一个消息队列。比如有以下的需求，一个子进程向队列中写数据，另外一个进程从队列中取数据的例子：
from multiprocessing import Process, Queueimport os, time, randomdef write(q): for value in ['A', 'B', 'C']: print 'Put {0} to queue...'.format(value) q.put(value) time.sleep(random.random())def read(q): while True: if not q.empty(): value = q.get(True) print 'Get {0} from queue.'.format(value) time.sleep(random.random()) else: breakif __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：
from multiprocessing import Process, Queueimport os, time, randomdef write(q): for value in ['A', 'B', 'C']: print 'Put {0} to queue...'.format(value) q.put(value) time.sleep(random.random())def read(q): while True: if not q.empty(): value = q.get(True) print 'Get {0} from queue.'.format(value) time.sleep(random.random()) else: breakif __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() if not q.empty(): print 'q is not empty...' else: print 'q is empty...' print 'OK' if not q.empty(): print 'q is not empty...' else: print 'q is empty...' print 'done...' 父进程与子进程共享内存
定义普通的变量，不能实现在父进程和子进程之间共享：
import multiprocessingfrom multiprocessing import Pooldef changevalue(n, a): n = 3.14 a[0] = 5if __name__ == '__main__': num = 0 arr = range(10) p = Pool() p1 = p.apply_async(changevalue, args=(num, arr)) p.close() p.join() print num print arr[:]
结果输出num的值还是在父进程中定义的0，arr的第一个元素值还是0。
使用multiprocessing创建共享对象:
import multiprocessingdef changevalue(n, a): n.value = 3.14 a[0] = 5if __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() print num.value print arr[:] 结果输出num的值是在子进程中修改的3.14，arr的第一个元素值更改为5。
共享内存在Pool中的使用：
import multiprocessingfrom multiprocessing import Pooldef changevalue(n, a): n.value = 3.14 a[0] = 5if __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() print num.value print arr[:] 希望本文所述对大家Python程序设计有所帮助。
注：相关教程知识阅读请移步到python教程频道。
