怎么在Python中使用multiprocessing實現(xiàn)多進程

怎么在Python中使用multiprocessing實現(xiàn)多進程？相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

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

multprocessing模塊的核心就是使管理進程像管理線程一樣方便，每個進程有自己獨立的GIL，所以不存在進程間爭搶GIL的問題，在多核CPU環(huá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是一個類，用于創(chuàng)建進程，以及定義進程的方法，Process類的構(gòu)造函數(shù)是：

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

參數(shù)含義：

• group：進程所屬組，基本不用

• target： 創(chuàng)建進程關(guān)聯(lián)的對象，需要傳入要多進程處理的函數(shù)名

• name： 定義進程的名稱

• args： 表示與target相關(guān)聯(lián)的函數(shù)的傳入?yún)?shù)，可以傳入多個，注意args是一個元組，如果傳入的參數(shù)只有一個，需要表示為 args = (element1,)

• kwargs： 表示調(diào)用對象的字典

程序解讀：

• multiprocessing.cpu_count()： 返回機器上cpu核的總數(shù)量

• p1.daemon = False ： 定義子進程的運行屬性，如果 .daemon設(shè)置為False表示子進程可以在主進程完成之后繼續(xù)執(zhí)行； 如果 .daemon設(shè)置為True，表示子進程隨著主進程的結(jié)束而結(jié)束；必須在start之前設(shè)置；

• p1.start(): 開始執(zhí)行子進程p1

• p1.join(): 定義子進程p1的運行阻塞主進程，只有p1子進程執(zhí)行完成之后才會繼續(xù)執(zhí)行join之后的主進程，但是子進程間互相不受join影響。

• 可以定義子進程阻塞主進程的時間--p1.join(100)，超時之后，主進程不再等待，開始執(zhí)行。join()需要放在start()方法之后；

• p1.terminate()：終止子進程的執(zhí)行，其后要跟上jion()方法更新子進程的狀態(tài)；

• p1.exitcode： 進程的退出狀態(tài)：  == 0 未生成任何錯誤，正常退出；  > 0 進程有一個錯誤，并以該錯誤碼退出；    <0 進程由一個-1 * exitcode信號結(jié)束

在multiprocessing中使用pool

如果需要多個子進程時，使用進程池(pool)來（自動）管理各個子進程更加方便：

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

• os.getpid()是獲取當前執(zhí)行的進程的ID

• p.close()方法是關(guān)掉進程池，表示不能再繼續(xù)向進程池添加進程了。

• p.join()方法是子進程阻塞主進程，必須在調(diào)用p.close()關(guān)閉進程池之后才能調(diào)用join()方法

多個子進程間的通信

多個子進程間的通信要用到multiprocessing.Queue，Queue的特性是它是一個消息隊列。比如有以下的需求，一個子進程向隊列中寫數(shù)據(jù)，另外一個進程從隊列中取數(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()：返回當前隊列包含的消息數(shù)量；

• Queue.empty()：如果隊列為空，返回True，反之False ；

• Queue.full()：如果隊列滿了，返回True,反之False；

• Queue.get():獲取隊列中的一條消息，然后將其從列隊中移除，可傳參超時時長；

• Queue.get_nowait()：相當Queue.get(False),取不到值時觸發(fā)異常：Empty；

• Queue.put():將一個值添加進數(shù)列，可傳參超時時長；

• Queue.put_nowait():相當于Queue.get(False),當隊列滿了時報錯：Full；

在進程池Pool中，使用Queue會出錯，需要使用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...'

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

定義普通的變量，不能實現(xià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的值還是在父進程中定義的0，arr的第一個元素值還是0。

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

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的值是在子進程中修改的3.14，arr的第一個元素值更改為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[:]

看完上述內(nèi)容，你們掌握怎么在Python中使用multiprocessing實現(xiàn)多進程的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！