一 multiprocessing模塊介紹
python中的多線程沒法利用多核優點，若是想要充分地使用多核CPU的資源（os.cpu_count()查看），在python中大部分狀況須要使用多進程。Python提供了multiprocessing。
multiprocessing模塊用來開啓子進程，並在子進程中執行咱們定製的任務（好比函數），該模塊與多線程模塊threading的編程接口相似。python
multiprocessing模塊的功能衆多：支持子進程、通訊和共享數據、執行不一樣形式的同步，提供了Process、Queue、Pipe、Lock等組件。git
須要再次強調的一點是：與線程不一樣，進程沒有任何共享狀態，進程修改的數據，改動僅限於該進程內。github
二 Process類的介紹
建立進程的類：數據庫
Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化獲得的對象，表示一個子進程中的任務（還沒有啓動） 強調： 1. 須要使用關鍵字的方式來指定參數 2. args指定的爲傳給target函數的位置參數，是一個元組形式，必須有逗號
參數介紹：編程
1 group參數未使用，值始終爲None
2
3 target表示調用對象，即子進程要執行的任務
4
5 args表示調用對象的位置參數元組，args=(1,2,'egon',)
6
7 kwargs表示調用對象的字典,kwargs={'name':'egon','age':18}
8
9 name爲子進程的名稱
方法介紹：json
1 p.start()：啓動進程，並調用該子進程中的p.run() 2 p.run():進程啓動時運行的方法，正是它去調用target指定的函數，咱們自定義類的類中必定要實現該方法 3 4 p.terminate():強制終止進程p，不會進行任何清理操做，若是p建立了子進程，該子進程就成了殭屍進程，使用該方法須要特別當心這種狀況。若是p還保存了一個鎖那麼也將不會被釋放，進而致使死鎖 5 p.is_alive():若是p仍然運行，返回True 6 7 p.join([timeout]):主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）。timeout是可選的超時時間，須要強調的是，p.join只能join住start開啓的進程，而不能join住run開啓的進程
屬性介紹：windows
1 p.daemon：默認值爲False，若是設爲True，表明p爲後臺運行的守護進程，當p的父進程終止時，p也隨之終止，而且設定爲True後，p不能建立本身的新進程，必須在p.start()以前設置 2 3 p.name:進程的名稱 4 5 p.pid：進程的pid 6 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.須要本身加鎖處理 #所以咱們最好找尋一種解決方案可以兼顧：一、效率高（多個進程共享一塊內存的數據）二、幫咱們處理好鎖問題。這就是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異常. 3 4 q.get_nowait():同q.get(False) 5 q.put_nowait():同q.put(False) 6 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 objects and allows 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, Value and 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被垃圾回收，將自動調用此函數
應用：
from multiprocessing import Pool import os,time def work(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)
from multiprocessing import Pool import os,time def work(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 in res_l: print(res.get()) #使用get來獲取apply_aync的結果,若是是apply,則沒有get方法,由於apply是同步執行,馬上獲取結果,也根本無需get
#一：使用進程池（異步調用,apply_async） #coding: utf-8 from multiprocessing import Process,Pool import time def func(msg): print( "msg:", msg) time.sleep(1) return msg if __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) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>對象組成的列表,而非最終的結果,但這一步是在join後執行的,證實結果已經計算完畢,剩下的事情就是調用每一個對象下的get方法去獲取結果 for i in res_l: print(i.get()) #使用get來獲取apply_aync的結果,若是是apply,則沒有get方法,由於apply是同步執行,馬上獲取結果,也根本無需get #二：使用進程池（同步調用,apply） #coding: utf-8 from multiprocessing import Process,Pool import time def func(msg): print( "msg:", msg) time.sleep(0.1) return msg if __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)
練習2：使用進程池維護固定數目的進程（重寫練習1）
#Pool內的進程數默認是cpu核數，假設爲4（查看方法os.cpu_count()） #開啓6個客戶端，會發現2個客戶端處於等待狀態 #在每一個進程內查看pid，會發現pid使用爲4個，即多個客戶端公用4個進程 from socket import * from multiprocessing import Pool import os server=socket(AF_INET,SOCK_STREAM) server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) server.bind(('127.0.0.1',8080)) server.listen(5) def talk(conn,client_addr): print('進程pid: %s' %os.getpid()) while True: try: msg=conn.recv(1024) if not msg:break conn.send(msg.upper()) except Exception: break if __name__ == '__main__': p=Pool() while True: conn,client_addr=server.accept() p.apply_async(talk,args=(conn,client_addr)) # p.apply(talk,args=(conn,client_addr)) #同步的話，則同一時間只有一個客戶端能訪問
from socket import * client=socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',8080)) while True: msg=input('>>: ').strip() if not msg:continue client.send(msg.encode('utf-8')) msg=client.recv(1024) print(msg.decode('utf-8'))
發現：併發開啓多個客戶端，服務端同一時間只有3個不一樣的pid，幹掉一個客戶端，另一個客戶端纔會進來，被3個進程之一處理
回掉函數：
須要回調函數的場景：進程池中任何一個任務一旦處理完了，就當即告知主進程：我好了額，你能夠處理個人結果了。主進程則調用一個函數去處理該結果，該函數即回調函數
咱們能夠把耗時間（阻塞）的任務放到進程池中，而後指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。
from multiprocessing import Pool import requests import json import os def get_page(url): print('<進程%s> get %s' %(os.getpid(),url)) respone=requests.get(url) if respone.status_code == 200: return {'url':url,'text':respone.text} def pasrse_page(res): print('<進程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> 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 in urls: 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的結果,其實徹底不必拿該結果,該結果已經傳給回調函數處理了 ''' 打印結果: <進程3388> get https://www.baidu.com <進程3389> get https://www.python.org <進程3390> get https://www.openstack.org <進程3388> get https://help.github.com/ <進程3387> parse https://www.baidu.com <進程3389> get http://www.sina.com.cn/ <進程3387> parse https://www.python.org <進程3387> parse https://help.github.com/ <進程3387> parse http://www.sina.com.cn/ <進程3387> parse https://www.openstack.org [{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}] '''
from multiprocessing import Pool import time,random import requests import re def get_page(url,pattern): response=requests.get(url) if response.status_code == 200: return (response.text,pattern) def parse_page(info): page_content,pattern=info res=re.findall(pattern,page_content) for item in res: dic={ 'index':item[0], 'title':item[1], 'actor':item[2].strip()[3:], 'time':item[3][5:], 'score':item[4]+item[5] } print(dic) if __name__ == '__main__': pattern1=re.compile(r'<dd>.*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S) url_dic={ 'http://maoyan.com/board/7':pattern1, } p=Pool() res_l=[] for url,pattern in url_dic.items(): res=p.apply_async(get_page,args=(url,pattern),callback=parse_page) res_l.append(res) for i in res_l: i.get() # res=requests.get('http://maoyan.com/board/7') # print(re.findall(pattern,res.text))
若是在主進程中等待進程池中全部任務都執行完畢後，再統一處理結果，則無需回調函數
from multiprocessing import Pool import time,random,os def work(n): time.sleep(1) return n**2 if __name__ == '__main__': p=Pool() res_l=[] for i in range(10): res=p.apply_async(work,args=(i,)) res_l.append(res) p.close() p.join() #等待進程池中全部進程執行完畢 nums=[] for res in res_l: nums.append(res.get()) #拿到全部結果 print(nums) #主進程拿到全部的處理結果,能夠在主進程中進行統一進行處理
進程池的其餘實現方式：https://docs.python.org/dev/library/concurrent.futures.html