python 多進程并發_python并發編程之多進程

一 multiprocessing模塊介紹

python中的多線程無法利用多核優勢，如果想要充分地使用多核CPU的資源(os.cpu_count()查看)，在python中大部分情況需要使用多進程。Python提供了multiprocessing。

multiprocessing模塊用來開啟子進程，并在子進程中執行我們定制的任務(比如函數)，該模塊與多線程模塊threading的編程接口類似。

multiprocessing模塊的功能眾多：支持子進程、通信和共享數據、執行不同形式的同步，提供了Process、Queue、Pipe、Lock等組件。

需要再次強調的一點是：與線程不同，進程沒有任何共享狀態，進程修改的數據，改動僅限于該進程內。

二 Process類的介紹

創建進程的類：

Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化得到的對象，表示一個子進程中的任務(尚未啟動)

強調：

1. 需要使用關鍵字的方式來指定參數

2. args指定的為傳給target函數的位置參數，是一個元組形式，必須有逗號

參數介紹：

1 group參數未使用，值始終為None

3 target表示調用對象，即子進程要執行的任務

5 args表示調用對象的位置參數元組，args=(1,2,'egon',)

7 kwargs表示調用對象的字典,kwargs={'name':'egon','age':18}

9 name為子進程的名稱

方法介紹：

1 p.start()：啟動進程，并調用該子進程中的p.run()

2 p.run():進程啟動時運行的方法，正是它去調用target指定的函數，我們自定義類的類中一定要實現該方法

4 p.terminate():強制終止進程p，不會進行任何清理操作，如果p創建了子進程，該子進程就成了僵尸進程，使用該方法需要特別小心這種情況。如果p還保存了一個鎖那么也將不會被釋放，進而導致死鎖

5 p.is_alive():如果p仍然運行，返回True

7 p.join([timeout]):主線程等待p終止(強調：是主線程處于等的狀態，而p是處于運行的狀態)。timeout是可選的超時時間，需要強調的是，p.join只能join住start開啟的進程，而不能join住run開啟的進程

屬性介紹：

1 p.daemon：默認值為False，如果設為True，代表p為后臺運行的守護進程，當p的父進程終止時，p也隨之終止，并且設定為True后，p不能創建自己的新進程，必須在p.start()之前設置2

3 p.name:進程的名稱4

5 p.pid：進程的pid6

7 p.exitcode:進程在運行時為None、如果為–N，表示被信號N結束(了解即可)8

9 p.authkey:進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是為涉及網絡連接的底層進程間通信提供安全性，這類連接只有在具有相同的身份驗證鍵時才能成功(了解即可)

三 Process類的使用

注意：在windows中Process()必須放到# if __name__ == '__main__':下

詳細解釋

創建并開啟子進程的兩種方式

方法一

方法二

進程直接的內存空間是隔離的

View Code

練習1：把上周所學的socket通信變成并發的形式

server端

多個client端

這么實現有沒有問題？？？

Process對象的join方法

join：主進程等，等待子進程結束

有了join，程序不就是串行了嗎？？？

Process對象的其他方法或屬性(了解)

terminate與is_alive

name與pid

僵尸進程與孤兒進程(了解)

View Code

四守護進程

主進程創建守護進程

其一：守護進程會在主進程代碼執行結束后就終止

其二：守護進程內無法再開啟子進程,否則拋出異常：AssertionError: daemonic processes are not allowed to have children

注意：進程之間是互相獨立的，主進程代碼運行結束，守護進程隨即終止

View Code

迷惑人的例子

五進程同步(鎖)

進程之間數據不共享,但是共享同一套文件系統,所以訪問同一個文件,或同一個打印終端,是沒有問題的,

而共享帶來的是競爭，競爭帶來的結果就是錯亂，如何控制，就是加鎖處理

part1：多個進程共享同一打印終端

并發運行,效率高,但競爭同一打印終端,帶來了打印錯亂

加鎖：由并發變成了串行,犧牲了運行效率,但避免了競爭

part2：多個進程共享同一文件

文件當數據庫，模擬搶票

并發運行，效率高，但競爭寫同一文件，數據寫入錯亂

加鎖：購票行為由并發變成了串行，犧牲了運行效率，但保證了數據安全

總結：

#加鎖可以保證多個進程修改同一塊數據時，同一時間只能有一個任務可以進行修改，即串行的修改，沒錯，速度是慢了，但犧牲了速度卻保證了數據安全。

雖然可以用文件共享數據實現進程間通信，但問題是：1.效率低(共享數據基于文件，而文件是硬盤上的數據)2.需要自己加鎖處理#因此我們最好找尋一種解決方案能夠兼顧：1、效率高(多個進程共享一塊內存的數據)2、幫我們處理好鎖問題。這就是mutiprocessing模塊為我們提供的基于消息的IPC通信機制：隊列和管道。

1隊列和管道都是將數據存放于內存中2 隊列又是基于(管道+鎖)實現的，可以讓我們從復雜的鎖問題中解脫出來，

我們應該盡量避免使用共享數據，盡可能使用消息傳遞和隊列，避免處理復雜的同步和鎖問題，而且在進程數目增多時，往往可以獲得更好的可獲展性。

六隊列(推薦使用)

進程彼此之間互相隔離，要實現進程間通信(IPC)，multiprocessing模塊支持兩種形式：隊列和管道，這兩種方式都是使用消息傳遞的

創建隊列的類(底層就是以管道和鎖定的方式實現)：

1 Queue([maxsize]):創建共享的進程隊列，Queue是多進程安全的隊列，可以使用Queue實現多進程之間的數據傳遞。

參數介紹：

1 maxsize是隊列中允許最大項數，省略則無大小限制。

方法介紹：

主要方法：

1 q.put方法用以插入數據到隊列中，put方法還有兩個可選參數：blocked和timeout。如果blocked為True(默認值)，并且timeout為正值，該方法會阻塞timeout指定的時間，直到該隊列有剩余的空間。如果超時，會拋出Queue.Full異常。如果blocked為False，但該Queue已滿，會立即拋出Queue.Full異常。

2 q.get方法可以從隊列讀取并且刪除一個元素。同樣，get方法有兩個可選參數：blocked和timeout。如果blocked為True(默認值)，并且timeout為正值，那么在等待時間內沒有取到任何元素，會拋出Queue.Empty異常。如果blocked為False，有兩種情況存在，如果Queue有一個值可用，則立即返回該值，否則，如果隊列為空，則立即拋出Queue.Empty異常.

4 q.get_nowait():同q.get(False)

5 q.put_nowait():同q.put(False)

7 q.empty():調用此方法時q為空則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中又加入了項目。

8 q.full()：調用此方法時q已滿則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中的項目被取走。

9 q.qsize():返回隊列中目前項目的正確數量，結果也不可靠，理由同q.empty()和q.full()一樣

其他方法(了解)：

1 q.cancel_join_thread():不會在進程退出時自動連接后臺線程。可以防止join_thread()方法阻塞

2 q.close():關閉隊列，防止隊列中加入更多數據。調用此方法，后臺線程將繼續寫入那些已經入隊列但尚未寫入的數據，但將在此方法完成時馬上關閉。如果q被垃圾收集，將調用此方法。關閉隊列不會在隊列使用者中產生任何類型的數據結束信號或異常。例如，如果某個使用者正在被阻塞在get()操作上，關閉生產者中的隊列不會導致get()方法返回錯誤。

3 q.join_thread()：連接隊列的后臺線程。此方法用于在調用q.close()方法之后，等待所有隊列項被消耗。默認情況下，此方法由不是q的原始創建者的所有進程調用。調用q.cancel_join_thread方法可以禁止這種行為

應用：

View Code

生產者消費者模型

在并發編程中使用生產者和消費者模式能夠解決絕大多數并發問題。該模式通過平衡生產線程和消費線程的工作能力來提高程序的整體處理數據的速度。

為什么要使用生產者和消費者模式

在線程世界里，生產者就是生產數據的線程，消費者就是消費數據的線程。在多線程開發當中，如果生產者處理速度很快，而消費者處理速度很慢，那么生產者就必須等待消費者處理完，才能繼續生產數據。同樣的道理，如果消費者的處理能力大于生產者，那么消費者就必須等待生產者。為了解決這個問題于是引入了生產者和消費者模式。

什么是生產者消費者模式

生產者消費者模式是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產者生產完數據之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列里取，阻塞隊列就相當于一個緩沖區，平衡了生產者和消費者的處理能力。

基于隊列實現生產者消費者模型

View Code

#生產者消費者模型總結

#程序中有兩類角色

一類負責生產數據(生產者)

一類負責處理數據(消費者)#引入生產者消費者模型為了解決的問題是：

平衡生產者與消費者之間的速度差#如何實現：

生產者-》隊列——》消費者#生產者消費者模型實現類程序的解耦和

此時的問題是主進程永遠不會結束，原因是：生產者p在生產完后就結束了，但是消費者c在取空了q之后，則一直處于死循環中且卡在q.get()這一步。

解決方式無非是讓生產者在生產完畢后，往隊列中再發一個結束信號，這樣消費者在接收到結束信號后就可以break出死循環

生產者在生產完畢后發送結束信號None

注意：結束信號None，不一定要由生產者發，主進程里同樣可以發，但主進程需要等生產者結束后才應該發送該信號

主進程在生產者生產完畢后發送結束信號None

但上述解決方式，在有多個生產者和多個消費者時，我們則需要用一個很low的方式去解決

有幾個消費者就需要發送幾次結束信號：相當low

其實我們的思路無非是發送結束信號而已，有另外一種隊列提供了這種機制

#JoinableQueue([maxsize])：這就像是一個Queue對象，但隊列允許項目的使用者通知生成者項目已經被成功處理。通知進程是使用共享的信號和條件變量來實現的。

#參數介紹：

maxsize是隊列中允許最大項數，省略則無大小限制。#方法介紹：

JoinableQueue的實例p除了與Queue對象相同的方法之外還具有：

q.task_done()：使用者使用此方法發出信號，表示q.get()的返回項目已經被處理。如果調用此方法的次數大于從隊列中刪除項目的數量，將引發ValueError異常

q.join():生產者調用此方法進行阻塞，直到隊列中所有的項目均被處理。阻塞將持續到隊列中的每個項目均調用q.task_done()方法為止

View Code

七管道

進程間通信(IPC)方式二：管道(不推薦使用，了解即可)

介紹

基于管道實現進程間通信(與隊列的方式是類似的，隊列就是管道加鎖實現的)

注意：生產者和消費者都沒有使用管道的某個端點，就應該將其關閉，如在生產者中關閉管道的右端，在消費者中關閉管道的左端。如果忘記執行這些步驟，程序可能再消費者中的recv()操作上掛起。管道是由操作系統進行引用計數的,必須在所有進程中關閉管道后才能生產EOFError異常。因此在生產者中關閉管道不會有任何效果，付費消費者中也關閉了相同的管道端點。

管道可以用于雙向通信，利用通常在客戶端/服務器中使用的請求／響應模型或遠程過程調用，就可以使用管道編寫與進程交互的程序

八共享數據

展望未來，基于消息傳遞的并發編程是大勢所趨

即便是使用線程，推薦做法也是將程序設計為大量獨立的線程集合

通過消息隊列交換數據。這樣極大地減少了對使用鎖定和其他同步手段的需求，

還可以擴展到分布式系統中

進程間通信應該盡量避免使用本節所講的共享數據的方式

進程間數據是獨立的，可以借助于隊列或管道實現通信，二者都是基于消息傳遞的

雖然進程間數據獨立，但可以通過Manager實現數據共享，事實上Manager的功能遠不止于此

A manager object returned by Manager() controls a server process which holds Python objectsandallows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Valueand Array. For example,

進程之間操作共享的數據

九信號量(了解)

信號量Semahpore(同線程一樣)

十事件(了解)

Event(同線程一樣)

十一進程池

在利用Python進行系統管理的時候，特別是同時操作多個文件目錄，或者遠程控制多臺主機，并行操作可以節約大量的時間。多進程是實現并發的手段之一，需要注意的問題是：

很明顯需要并發執行的任務通常要遠大于核數

一個操作系統不可能無限開啟進程，通常有幾個核就開幾個進程

進程開啟過多，效率反而會下降(開啟進程是需要占用系統資源的，而且開啟多余核數目的進程也無法做到并行)

例如當被操作對象數目不大時，可以直接利用multiprocessing中的Process動態成生多個進程，十幾個還好，但如果是上百個，上千個。。。手動的去限制進程數量卻又太過繁瑣，此時可以發揮進程池的功效。

我們就可以通過維護一個進程池來控制進程數目，比如httpd的進程模式，規定最小進程數和最大進程數...

ps：對于遠程過程調用的高級應用程序而言，應該使用進程池，Pool可以提供指定數量的進程，供用戶調用，當有新的請求提交到pool中時，如果池還沒有滿，那么就會創建一個新的進程用來執行該請求；但如果池中的進程數已經達到規定最大值，那么該請求就會等待，直到池中有進程結束，就重用進程池中的進程。

創建進程池的類：如果指定numprocess為3，則進程池會從無到有創建三個進程，然后自始至終使用這三個進程去執行所有任務，不會開啟其他進程

1 Pool([numprocess [,initializer [, initargs]]]):創建進程池

參數介紹：

1 numprocess:要創建的進程數，如果省略，將默認使用cpu_count()的值

2 initializer：是每個工作進程啟動時要執行的可調用對象，默認為None

3 initargs：是要傳給initializer的參數組

方法介紹：

主要方法：

1 p.apply(func [, args [, kwargs]]):在一個池工作進程中執行func(*args,**kwargs),然后返回結果。需要強調的是：此操作并不會在所有池工作進程中并執行func函數。如果要通過不同參數并發地執行func函數，必須從不同線程調用p.apply()函數或者使用p.apply_async()2 p.apply_async(func [, args [, kwargs]]):在一個池工作進程中執行func(*args,**kwargs),然后返回結果。此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變為可用時，將理解傳遞給callback。callback禁止執行任何阻塞操作，否則將接收其他異步操作中的結果。3

4 p.close():關閉進程池，防止進一步操作。如果所有操作持續掛起，它們將在工作進程終止前完成5 P.jion():等待所有工作進程退出。此方法只能在close()或teminate()之后調用

其他方法(了解部分)

方法apply_async()和map_async()的返回值是AsyncResul的實例obj。實例具有以下方法

obj.get():返回結果，如果有必要則等待結果到達。timeout是可選的。如果在指定時間內還沒有到達，將引發一場。如果遠程操作中引發了異常，它將在調用此方法時再次被引發。

obj.ready():如果調用完成，返回True

obj.successful():如果調用完成且沒有引發異常，返回True，如果在結果就緒之前調用此方法，引發異常

obj.wait([timeout]):等待結果變為可用。

obj.terminate()：立即終止所有工作進程，同時不執行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調用此函數

View Code

應用：

from multiprocessing importPoolimportos,timedefwork(n):print('%s run' %os.getpid())

time.sleep(3)return n**2

if __name__ == '__main__':

p=Pool(3) #進程池中從無到有創建三個進程,以后一直是這三個進程在執行任務

res_l=[]for i in range(10):

res=p.apply(work,args=(i,)) #同步調用，直到本次任務執行完畢拿到res，等待任務work執行的過程中可能有阻塞也可能沒有阻塞，但不管該任務是否存在阻塞，同步調用都會在原地等著，只是等的過程中若是任務發生了阻塞就會被奪走cpu的執行權限

res_l.append(res)print(res_l)

同步調用apply

from multiprocessing importPoolimportos,timedefwork(n):print('%s run' %os.getpid())

time.sleep(3)return n**2

if __name__ == '__main__':

p=Pool(3) #進程池中從無到有創建三個進程,以后一直是這三個進程在執行任務

res_l=[]for i in range(10):

res=p.apply_async(work,args=(i,)) #同步運行,阻塞、直到本次任務執行完畢拿到res

res_l.append(res)#異步apply_async用法：如果使用異步提交的任務，主進程需要使用jion，等待進程池內任務都處理完，然后可以用get收集結果，否則，主進程結束，進程池可能還沒來得及執行，也就跟著一起結束了

p.close()

p.join()for res inres_l:print(res.get()) #使用get來獲取apply_aync的結果,如果是apply,則沒有get方法,因為apply是同步執行,立刻獲取結果,也根本無需get

異步調用apply_async

#一：使用進程池(異步調用,apply_async)#coding: utf-8

from multiprocessing importProcess,Poolimporttimedeffunc(msg):print( "msg:", msg)

time.sleep(1)returnmsgif __name__ == "__main__":

pool= Pool(processes = 3)

res_l=[]for i in range(10):

msg= "hello %d" %(i)

res=pool.apply_async(func, (msg, )) #維持執行的進程總數為processes，當一個進程執行完畢后會添加新的進程進去

res_l.append(res)print("==============================>") #沒有后面的join，或get，則程序整體結束，進程池中的任務還沒來得及全部執行完也都跟著主進程一起結束了

pool.close()#關閉進程池，防止進一步操作。如果所有操作持續掛起，它們將在工作進程終止前完成

pool.join() #調用join之前，先調用close函數，否則會出錯。執行完close后不會有新的進程加入到pool,join函數等待所有子進程結束

print(res_l) #看到的是對象組成的列表,而非最終的結果,但這一步是在join后執行的,證明結果已經計算完畢,剩下的事情就是調用每個對象下的get方法去獲取結果

for i inres_l:print(i.get()) #使用get來獲取apply_aync的結果,如果是apply,則沒有get方法,因為apply是同步執行,立刻獲取結果,也根本無需get

#二：使用進程池(同步調用,apply)#coding: utf-8

from multiprocessing importProcess,Poolimporttimedeffunc(msg):print( "msg:", msg)

time.sleep(0.1)returnmsgif __name__ == "__main__":

pool= Pool(processes = 3)

res_l=[]for i in range(10):

msg= "hello %d" %(i)

res=pool.apply(func, (msg, )) #維持執行的進程總數為processes，當一個進程執行完畢后會添加新的進程進去

res_l.append(res) #同步執行，即執行完一個拿到結果，再去執行另外一個

print("==============================>")

pool.close()

pool.join()#調用join之前，先調用close函數，否則會出錯。執行完close后不會有新的進程加入到pool,join函數等待所有子進程結束

print(res_l) #看到的就是最終的結果組成的列表

for i in res_l: #apply是同步的，所以直接得到結果，沒有get()方法

print(i)

詳解：apply_async與apply

練習2：使用進程池維護固定數目的進程(重寫練習1)

#Pool內的進程數默認是cpu核數，假設為4(查看方法os.cpu_count())#開啟6個客戶端，會發現2個客戶端處于等待狀態#在每個進程內查看pid，會發現pid使用為4個，即多個客戶端公用4個進程

from socket import *

from multiprocessing importPoolimportos

server=socket(AF_INET,SOCK_STREAM)

server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)

server.bind(('127.0.0.1',8080))

server.listen(5)deftalk(conn,client_addr):print('進程pid: %s' %os.getpid())whileTrue:try:

msg=conn.recv(1024)if not msg:breakconn.send(msg.upper())exceptException:break

if __name__ == '__main__':

p=Pool()whileTrue:

conn,client_addr=server.accept()

p.apply_async(talk,args=(conn,client_addr))#p.apply(talk,args=(conn,client_addr)) #同步的話，則同一時間只有一個客戶端能訪問

server端

from socket import *client=socket(AF_INET,SOCK_STREAM)

client.connect(('127.0.0.1',8080))whileTrue:

msg=input('>>:').strip()if not msg:continueclient.send(msg.encode('utf-8'))

msg=client.recv(1024)print(msg.decode('utf-8'))

客戶端

發現：并發開啟多個客戶端，服務端同一時間只有3個不同的pid，干掉一個客戶端，另外一個客戶端才會進來，被3個進程之一處理

回掉函數：

需要回調函數的場景：進程池中任何一個任務一旦處理完了，就立即告知主進程：我好了額，你可以處理我的結果了。主進程則調用一個函數去處理該結果，該函數即回調函數

我們可以把耗時間(阻塞)的任務放到進程池中，然后指定回調函數(主進程負責執行)，這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。

from multiprocessing importPoolimportrequestsimportjsonimportosdefget_page(url):print(' get %s' %(os.getpid(),url))

respone=requests.get(url)if respone.status_code == 200:return {'url':url,'text':respone.text}defpasrse_page(res):print(' parse %s' %(os.getpid(),res['url']))

parse_res='url: size:[%s]\n' %(res['url'],len(res['text']))

with open('db.txt','a') as f:

f.write(parse_res)if __name__ == '__main__':

urls=['https://www.baidu.com','https://www.python.org','https://www.openstack.org','https://help.github.com/','http://www.sina.com.cn/']

p=Pool(3)

res_l=[]for url inurls:

res=p.apply_async(get_page,args=(url,),callback=pasrse_page)

res_l.append(res)

p.close()

p.join()print([res.get() for res in res_l]) #拿到的是get_page的結果,其實完全沒必要拿該結果,該結果已經傳給回調函數處理了

'''打印結果:

get https://www.baidu.com