[TOC] ## **進程池** ### **為什么要有進程池** 在程序實際處理問題過程中，忙時會有成千上萬的任務需要被執行，閑時可能只有零星任務。那么在成千上萬個任務需要被執行的時候，我們就需要去創建成千上萬個進程么？ 首先，創建進程需要消耗時間，銷毀進程也需要消耗時間。 第二即便開啟了成千上萬的進程，操作系統也不能讓他們同時執行，這樣反而會影響程序的效率。 因此我們不能無限制的根據任務開啟或者結束進程。 ### **進程池的概念** 定義一個池子，在里面放上固定數量的進程，有需求來了，就拿一個池中的進程來處理任務，等到處理完畢，進程并不關閉，而是將進程再放回進程池中繼續等待任務。 如果有很多任務需要執行，池中的進程數量不夠，任務就要等待之前的進程執行任務完畢歸來，拿到空閑進程才能繼續執行。 就是說，池中進程的數量是固定的，那么同一時間最多有固定數量的進程在運行。這樣不會增加操作系統的調度難度，還節省了開閉進程的時間，也一定程度上能夠實現并發效果。 ### 多進程和進程池的對比 對于純計算型的代碼 使用進程池更好 —— 真理 對于高IO的代碼 直接使用多進程更好 —— 相對論 **結論:** 進程池比起多進程來說 節省了開啟進程回收進程資源的時間,給操作系統調度進程降低了難度 ## **用法語法** Pool([numprocess [,initializer [, initargs]]]):創建進程池 ### **參數介紹** 1. numprocess: 要創建的進程數，如果省略，將默認使用cpu_count()的值 2. initializer： 是每個工作進程啟動時要執行的可調用對象，默認為None 3. initargs： 是要傳給initializer的參數組 ### **主要方法** 1. p.apply(func [, args [, kwargs]]): 在一個池工作進程中執行`func(*args,**kwargs)`,然后返回結果。(**同步**) 2. p.apply_async(func [, args [, kwargs]]): 在一個池工作進程中執行`func(*args,**kwargs)`,然后返回結果。(**異步**) 回調函數后面講 3. p.close(): 關閉進程池，防止進一步操作。 4. p.jion(): 等待所有工作進程退出。此方法只能在close（）或teminate()之后調用 5. p.map() 接收一個任務函數，和一個iterable。節省了for循環和close、join，是一種簡便的寫法 ### 其他方法(了解) 方法apply_async和map_async的返回值是AsyncResul的實例obj。實例具有以下方法 1. obj.get(): 返回結果，如果有必要則等待結果到達。 timeout是可選的。如果在指定時間內還沒有到達，將引發異常。 如果遠程操作中引發了異常，它將在調用此方法時再次被引發。 2. obj.ready(): 如果調用完成，返回True 3. obj.successful(): 如果調用完成且沒有引發異常，返回True，如果在結果就緒之前調用此方法，引發異常 4. obj.wait([timeout]): 等待結果變為可用。 5. obj.terminate()： 立即終止所有工作進程，同時不執行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調用此函數 ## 代碼實例 ### apply_async代碼舉例 ~~~ from multiprocessing import Pool import os,time,random def work(i): print('%s is 運行,PID:%s'%(i,os.getpid())) time.sleep(random.random()) return i**i if __name__ == '__main__': p=Pool(3) #進程數量為3的進程池 res_1=[] for i in range(5): res=p.apply_async(func=work,args=(i,)) res_1.append(res) p.close() p.join() print('end'.center(15,'-')) for r in res_1: print(r.get()) ~~~ 說明: ~~~ # 異步運行，根據進程池中有的進程數，每次最多3個子進程在異步執行 # 返回結果之后，將結果放入列表，歸還進程，之后再執行新的任務 # 需要注意的是，進程池中的三個進程不會同時開啟或者同時結束 # 而是執行完一個就釋放一個進程，這個進程就去接收新的任務。 ~~~ 運行結果: ``` 0 is 運行,PID:5732 1 is 運行,PID:5252 2 is 運行,PID:6472 3 is 運行,PID:5732 4 is 運行,PID:6472 ------end------ 1 1 4 27 256 ``` ### map代碼舉例 以下兩段代碼完全等價,map幫我們完成了for循環,close,join任務 ~~~ p = Pool(5) for i in range(10): p.apply_async(func=wahaha,args=(i,)) # 異步提交了一個任務 p.close() p.join() ~~~ ~~~ p = Pool(5) p.map(func=wahaha,iterable=range(10)) ~~~ apply_async和map相比，操作復雜，但是可以通過get方法獲取返回值 ## **回調函數** ### 需要回調函數的場景： 進程池中任何一個任務一旦處理完了，就立即告知主進程：我好了，你可以處理我的結果了。主進程則調用一個函數去處理該結果，該函數即回調函數 我們可以把耗時間（阻塞）的任務放到進程池中，然后指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。 ### 代碼舉例 ~~~ from multiprocessing import Pool import os,time,random def work1(i): print('%s is 運行,PID:%s'%(i,os.getpid())) time.sleep(random.random()) return i*i def work2(f): # 回調函數用的是主進程中的資源 print(f,os.getpid()) if __name__ == '__main__': p=Pool(3) #進程數量為3的進程池 res_1=[] print('主進程:%s'%os.getpid()) p.apply_async(func=work1,args=(9,),callback=work2) # callback是一個回調函數，接收一個函數地址 p.close() p.join() ~~~ 執行結果: ``` 主進程:8080 9 is 運行,PID:2988 81 8080 ```