前言：

python多進(jìn)程，經(jīng)常在使用，卻沒(méi)有怎么系統(tǒng)的學(xué)習(xí)過(guò)，官網(wǎng)上面講得比較細(xì)，結(jié)合自己的學(xué)習(xí)，整理記錄下官網(wǎng):https://docs.python.org/3/library/multiprocessing.html

multiprocessing簡(jiǎn)介

multiprocessing是python自帶的多進(jìn)程模塊，可以大批量的生成進(jìn)程，在服務(wù)器為多核CPU時(shí)效果更好，類(lèi)似于threading模塊。相對(duì)于多線程，多進(jìn)程由于獨(dú)享內(nèi)存空間，更穩(wěn)定安全，在運(yùn)維里面做些批量操作時(shí)，多進(jìn)程有更多適用的場(chǎng)景

multiprocessing包提供了本地和遠(yuǎn)程兩種并發(fā)操作,有效的避開(kāi)了使用子進(jìn)程而不是全局解釋鎖的線程，因此，multiprocessing可以有效利用到多核處理

Process類(lèi)

在multiporcessing中，通過(guò)Process類(lèi)對(duì)象來(lái)批量產(chǎn)生進(jìn)程，使用start()方法來(lái)啟動(dòng)這個(gè)進(jìn)程

1.語(yǔ)法

multiprocessing.Process(group=None,target=None,name=None,args=(),kwargs={},*)

• group: 這個(gè)參數(shù)一般為空，它只是為了兼容threading.Tread
• target: 這個(gè)參數(shù)就是通過(guò)run()可調(diào)用對(duì)象的方法，默認(rèn)為空，表示沒(méi)有方法被調(diào)用
• name: 表示進(jìn)程名
• args: 傳給target調(diào)用方法的tuple(元組)參數(shù)
• kwargs: 傳給target調(diào)用方法的dict(字典)參數(shù)

2.Process類(lèi)的方法及對(duì)象

run()
該方法是進(jìn)程的運(yùn)行過(guò)程，可以在子類(lèi)中重寫(xiě)此方法，一般也很少去重構(gòu)

start()
啟動(dòng)進(jìn)程，每個(gè)進(jìn)程對(duì)象都必須被該方法調(diào)用

join([timeout])
等待進(jìn)程終止，再往下執(zhí)行，可以設(shè)置超時(shí)時(shí)間

name
可以獲取進(jìn)程名字,多個(gè)進(jìn)程也可以是相同的名字

is_alive()
返回進(jìn)程是否還存活，True or False，進(jìn)程存活是指start()開(kāi)始到子進(jìn)程終止

daemon
守護(hù)進(jìn)程的標(biāo)記，一個(gè)布爾值，在start()之后設(shè)置該值，表示是否后臺(tái)運(yùn)行
注意：如果設(shè)置了后臺(tái)運(yùn)行，那么后臺(tái)程序不運(yùn)行再創(chuàng)建子進(jìn)程

pid
可以獲取進(jìn)程ID

exitcode
子進(jìn)程退出時(shí)的值，如果進(jìn)程還沒(méi)有終止，值將是None,如果是負(fù)值，表示子進(jìn)程被終止

terminate()
終止進(jìn)程，如果是Windows，則使用terminateprocess()，該方法對(duì)已經(jīng)退出和結(jié)束的進(jìn)程，將不會(huì)執(zhí)行

以下為一個(gè)簡(jiǎn)單的例子:

#-*- coding:utf8 -*- 
import multiprocessing
import time

def work(x):
  time.sleep(1)
  print time.ctime(),'這是子進(jìn)程[{0}]...'.format(x)

if __name__ == '__main__':
  for i in range(5):
    p = multiprocessing.Process(target=work,args=(i,))
    print '啟動(dòng)進(jìn)程數(shù):{0}'.format(i)
    p.start()
    p.deamon = True

當(dāng)然也可以顯示每個(gè)進(jìn)程的ID

#-*- coding:utf8 -*- 
import multiprocessing
import time
import os

def work(x):
  time.sleep(1)
  ppid = os.getppid()
  pid = os.getpid()
  print time.ctime(),'這是子進(jìn)程[{0},父進(jìn)程:{1},子進(jìn)程:{2}]...'.format(x,ppid,pid)

if __name__ == '__main__':
  for i in range(5):
    p = multiprocessing.Process(target=work,args=(i,))
    print '啟動(dòng)進(jìn)程數(shù):{0}'.format(i)
    p.start()
    p.deamon = True

但在實(shí)際使用的過(guò)程中，并不只是并發(fā)完就可以了，比如，有30個(gè)任務(wù)，由于服務(wù)器資源有限，每次并發(fā)5個(gè)任務(wù)，這里還涉及到30個(gè)任務(wù)怎么獲取的問(wèn)題，另外并發(fā)的進(jìn)程任務(wù)執(zhí)行時(shí)間很難保證一致，尤其是需要時(shí)間的任務(wù)，可能并發(fā)5個(gè)任務(wù)，有3個(gè)已經(jīng)執(zhí)行完了，2個(gè)還需要很長(zhǎng)時(shí)間執(zhí)行，總不能等到這兩個(gè)進(jìn)程執(zhí)行完了，再繼續(xù)執(zhí)行后面的任務(wù)，因此進(jìn)程控制就在此有了使用場(chǎng)景，可以利用Process的方法和一些multiprocessing的包，類(lèi)等結(jié)合使用

進(jìn)程控制及通信常用類(lèi)

一、Queue類(lèi)

類(lèi)似于python自帶的Queue.Queue，主要用在比較小的隊(duì)列上面

語(yǔ)法：

multiprocessing.Queue([maxsize])

類(lèi)方法：

qsize()
返回隊(duì)列的大致大小，因?yàn)槎噙M(jìn)程或者多線程一直在消耗隊(duì)列，因此該數(shù)據(jù)不一定正確

empty()
判斷隊(duì)列是否為空，如果是，則返回True，否則False

full()
判斷隊(duì)列是否已滿(mǎn)，如果是，則返回True，否則False

put(obj[, block[, timeout]])
將對(duì)象放入隊(duì)列，可選參數(shù)block為T(mén)rue，timeout為None

get()
從隊(duì)列取出對(duì)象

#-*- coding:utf8 -*-
from multiprocessing import Process, Queue

def f(q):
  q.put([42,None,'hi'])

if __name__ == '__main__':
  q = Queue()
  p = Process(target=f, args=(q,))
  p.start()
  print q.get() #打印內(nèi)容: [42,None,'hi']
  p.join()

二、Pipe類(lèi)

pipe()函數(shù)返回一對(duì)對(duì)象的連接，可以為進(jìn)程間傳輸消息，在打印一些日志、進(jìn)程控制上面有一些用處，Pip()對(duì)象返回兩個(gè)對(duì)象connection，代表兩個(gè)通道，每個(gè)connection對(duì)象都有send()和recv()方法，需要注意的是兩個(gè)或以上的進(jìn)程同時(shí)讀取或者寫(xiě)入同一管道，可能會(huì)導(dǎo)致數(shù)據(jù)混亂，測(cè)試了下，是直接覆蓋了。另外，返回的兩個(gè)connection,如果一個(gè)是send()數(shù)據(jù),那么另外一個(gè)就只能recv()接收數(shù)據(jù)了

#-*- coding:utf8 -*-
from multiprocessing import Process, Pipe
import time
def f(conn,i):
  print '[{0}]已經(jīng)執(zhí)行到子進(jìn)程:{1}'.format(time.ctime(),i)
  time.sleep(1)
  w = "[{0}]hi,this is :{1}".format(time.ctime(),i)
  conn.send(w)
  conn.close()

if __name__ == '__main__':
  reader = []
  parent_conn, child_conn = Pipe()
  for i in range(4):
    p = Process(target=f, args=(child_conn,i))
    p.start()
    reader.append(parent_conn)
    p.deamon=True

  # 等待所有子進(jìn)程跑完
  time.sleep(3)
  print '\n[{0}]下面打印child_conn向parent_conn傳輸?shù)男畔?'.format(time.ctime())
  for i in reader:
    print i.recv()

輸出為：

三、Value,Array

在進(jìn)行并發(fā)編程時(shí)，應(yīng)盡量避免使用共享狀態(tài)，因?yàn)槎噙M(jìn)程同時(shí)修改數(shù)據(jù)會(huì)導(dǎo)致數(shù)據(jù)破壞。但如果確實(shí)需要在多進(jìn)程間共享數(shù)據(jù)，multiprocessing也提供了方法Value、Array

from multiprocessing import Process, Value, Array

def f(n, a):
  n.value = 3.1415927
  for i in range(len(a)):
    a[i] = -a[i]

if __name__ == '__main__':
  num = Value('d',0.0)
  arr = Array('i', range(10))

  p = Process(target=f, args=(num, arr))
  p.start()
  p.join()

  print num.value
  print arr[:]

*print
3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]*

四、Manager進(jìn)程管理模塊

Manager類(lèi)管理進(jìn)程使用得較多，它返回對(duì)象可以操控子進(jìn)程，并且支持很多類(lèi)型的操作，如: list, dict, Namespace、lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value, Array，因此使用Manager基本上就夠了

from multiprocessing import Process, Manager

def f(d, l):
  d[1] = '1'
  d['2'] = 2
  d[0.25] = None
  l.reverse()

if __name__ == '__main__':
  with Manager() as manager:
    d = manager.dict()
    l = manager.list(range(10))

    p = Process(target=f, args=(d, l))
    p.start()
    p.join() #等待進(jìn)程結(jié)束后往下執(zhí)行
    print d,'\n',l

輸出：
{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

可以看到，跟共享數(shù)據(jù)一樣的效果，大部分管理進(jìn)程的方法都集成到了Manager()模塊了

五、對(duì)多進(jìn)程控制的應(yīng)用實(shí)例

  #-*- coding:utf8 -*-
  from multiprocessing import Process, Queue
  import time
  
  def work(pname,q):
    time.sleep(1)
    print_some = "{0}|this is process: {1}".format(time.ctime(),pname)
    print print_some
    q.put(pname)
  
  if __name__ == '__main__':
    p_manag_num = 2 # 進(jìn)程并發(fā)控制數(shù)量2
    # 并發(fā)的進(jìn)程名
    q_process = ['process_1','process_2','process_3','process_4','process_5']
    q_a = Queue() # 將進(jìn)程名放入隊(duì)列
    q_b = Queue() # 將q_a的進(jìn)程名放往q_b進(jìn)程,由子進(jìn)程完成
  
    for i in q_process:
      q_a.put(i)
  
    p_list = [] # 完成的進(jìn)程隊(duì)列
    while not q_a.empty():
      if len(p_list) <= 2:
        pname=q_a.get()
        p = Process(target=work, args=(pname,q_b))
        p.start()
        p_list.append(p)
        print pname
  
      for p in p_list:
        if not p.is_alive():
          p_list.remove(p)
  
    # 等待5秒,預(yù)估執(zhí)行完后看隊(duì)列通信信息
    # 當(dāng)然也可以循環(huán)判斷隊(duì)列里面的進(jìn)程是否執(zhí)行完成
    time.sleep(5)
    print '打印p_b隊(duì)列:'
    while not q_b.empty():
      print q_b.get()

執(zhí)行結(jié)果:

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論