快捷導(dǎo)航

Python multiprocessing多進(jìn)程原理與應(yīng)用示例

更新時(shí)間：2019年02月28日 14:30:17 作者：-牧野-

這篇文章主要介紹了Python multiprocessing多進(jìn)程原理與應(yīng)用,結(jié)合實(shí)例形式詳細(xì)分析了基于multiprocessing包的多進(jìn)程概念、原理及相關(guān)使用操作技巧,需要的朋友可以參考下

本文實(shí)例講述了Python multiprocessing多進(jìn)程原理與應(yīng)用。分享給大家供大家參考，具體如下：

multiprocessing包是Python中的多進(jìn)程管理包,可以利用multiprocessing.Process對(duì)象來(lái)創(chuàng)建進(jìn)程,Process對(duì)象擁有is_alive()、join([timeout])、run()、start()、terminate()等方法。

multprocessing模塊的核心就是使管理進(jìn)程像管理線程一樣方便，每個(gè)進(jìn)程有自己獨(dú)立的GIL，所以不存在進(jìn)程間爭(zhēng)搶GIL的問(wèn)題，在多核CPU環(huán)境中，可以大大提高運(yùn)行效率。

multiprocessing使用示例：

import multiprocessing
import time
import cv2
def 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是一個(gè)類，用于創(chuàng)建進(jìn)程，以及定義進(jìn)程的方法，Process類的構(gòu)造函數(shù)是：

def __init__(self, group=None, target=None, name=None, args=(), kwargs={})

參數(shù)含義：

group：進(jìn)程所屬組，基本不用
target：創(chuàng)建進(jìn)程關(guān)聯(lián)的對(duì)象，需要傳入要多進(jìn)程處理的函數(shù)名
name：定義進(jìn)程的名稱
args：表示與target相關(guān)聯(lián)的函數(shù)的傳入?yún)?shù)，可以傳入多個(gè)，注意args是一個(gè)元組，如果傳入的參數(shù)只有一個(gè)，需要表示為 args = (element1,)
kwargs：表示調(diào)用對(duì)象的字典

程序解讀：

multiprocessing.cpu_count()：返回機(jī)器上cpu核的總數(shù)量
p1.daemon = False ：定義子進(jìn)程的運(yùn)行屬性，如果 .daemon設(shè)置為False表示子進(jìn)程可以在主進(jìn)程完成之后繼續(xù)執(zhí)行；如果 .daemon設(shè)置為True，表示子進(jìn)程隨著主進(jìn)程的結(jié)束而結(jié)束；必須在start之前設(shè)置；
p1.start(): 開始執(zhí)行子進(jìn)程p1
p1.join(): 定義子進(jìn)程p1的運(yùn)行阻塞主進(jìn)程，只有p1子進(jìn)程執(zhí)行完成之后才會(huì)繼續(xù)執(zhí)行join之后的主進(jìn)程，但是子進(jìn)程間互相不受join影響。
可以定義子進(jìn)程阻塞主進(jìn)程的時(shí)間--p1.join(100)，超時(shí)之后，主進(jìn)程不再等待，開始執(zhí)行。join()需要放在start()方法之后；
p1.terminate()：終止子進(jìn)程的執(zhí)行，其后要跟上jion()方法更新子進(jìn)程的狀態(tài)；
p1.exitcode：進(jìn)程的退出狀態(tài)： == 0 未生成任何錯(cuò)誤，正常退出； > 0 進(jìn)程有一個(gè)錯(cuò)誤，并以該錯(cuò)誤碼退出； <0 進(jìn)程由一個(gè)-1 * exitcode信號(hào)結(jié)束

在multiprocessing中使用pool

如果需要多個(gè)子進(jìn)程時(shí)，使用進(jìn)程池(pool)來(lái)（自動(dòng)）管理各個(gè)子進(jìn)程更加方便：

from multiprocessing import Pool
import os, time
def 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類創(chuàng)建進(jìn)程的方法不同，Pool是通過(guò)apply_async(func,args=(args))方法創(chuàng)建進(jìn)程，一個(gè)進(jìn)程池中能同時(shí)運(yùn)行的任務(wù)數(shù)是機(jī)器上CPU核的總數(shù)量n_kernel，如果創(chuàng)建的子進(jìn)程數(shù)大于n_kernel，則同時(shí)執(zhí)行n_kernel個(gè)進(jìn)程，這n_kernel中某個(gè)進(jìn)程完成之后才會(huì)啟動(dòng)下一個(gè)進(jìn)程。

os.getpid()是獲取當(dāng)前執(zhí)行的進(jìn)程的ID
p.close()方法是關(guān)掉進(jìn)程池，表示不能再繼續(xù)向進(jìn)程池添加進(jìn)程了。
p.join()方法是子進(jìn)程阻塞主進(jìn)程，必須在調(diào)用p.close()關(guān)閉進(jìn)程池之后才能調(diào)用join()方法

多個(gè)子進(jìn)程間的通信

多個(gè)子進(jìn)程間的通信要用到multiprocessing.Queue，Queue的特性是它是一個(gè)消息隊(duì)列。比如有以下的需求，一個(gè)子進(jìn)程向隊(duì)列中寫數(shù)據(jù)，另外一個(gè)進(jìn)程從隊(duì)列中取數(shù)據(jù)的例子：

from multiprocessing import Process, Queue
import os, time, random
def 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:
      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()：返回當(dāng)前隊(duì)列包含的消息數(shù)量；
Queue.empty()：如果隊(duì)列為空，返回True，反之False ；
Queue.full()：如果隊(duì)列滿了，返回True,反之False；
Queue.get():獲取隊(duì)列中的一條消息，然后將其從列隊(duì)中移除，可傳參超時(shí)時(shí)長(zhǎng)；
Queue.get_nowait()：相當(dāng)Queue.get(False),取不到值時(shí)觸發(fā)異常：Empty；
Queue.put():將一個(gè)值添加進(jìn)數(shù)列，可傳參超時(shí)時(shí)長(zhǎng)；
Queue.put_nowait():相當(dāng)于Queue.get(False),當(dāng)隊(duì)列滿了時(shí)報(bào)錯(cuò)：Full；

在進(jìn)程池Pool中，使用Queue會(huì)出錯(cuò)，需要使用Manager.Queue：

from multiprocessing import Process, Queue
import os, time, random
def 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:
      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()
  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...'

父進(jìn)程與子進(jìn)程共享內(nèi)存

定義普通的變量，不能實(shí)現(xiàn)在父進(jìn)程和子進(jìn)程之間共享：

import multiprocessing
from multiprocessing import Pool
def changevalue(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()
  print num
  print arr[:]

結(jié)果輸出num的值還是在父進(jìn)程中定義的0，arr的第一個(gè)元素值還是0。

使用multiprocessing創(chuàng)建共享對(duì)象:

import multiprocessing
def changevalue(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()
  print num.value
  print arr[:]

結(jié)果輸出num的值是在子進(jìn)程中修改的3.14，arr的第一個(gè)元素值更改為5。

共享內(nèi)存在Pool中的使用：

import multiprocessing
from multiprocessing import Pool
def changevalue(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()
  print num.value
  print arr[:]

希望本文所述對(duì)大家Python程序設(shè)計(jì)有所幫助。

您可能感興趣的文章: