[TOC] ## 進程池與線程池統一的方法 我們知道multiprocessing 線程模塊里面有線程池,但是由于threading模塊被開發得較早,所以里面沒有池的概念和方法 后來python官方推出了包含的進程池和線程池等和池有關的所有工具的全新方法**concurrent.futures模塊** ### **concurrent.futures模塊介紹** 官網：https://docs.python.org/dev/library/concurrent.futures.html concurrent.futures模塊提供了高度封裝的異步調用接口,線程池和進程池使用方法一致 * ThreadPoolExecutor：線程池，提供異步調用 * ProcessPoolExecutor: 進程池，提供異步調用 ### **基本方法** 1. submit(fn, *args, **kwargs) 異步提交任務 2. map(func, *iterables, timeout=None, chunksize=1) 取代for循環submit的操作 3. shutdown(wait=True) 相當于進程池的pool.close()+pool.join()操作 wait=True，等待池內所有任務執行完畢回收完資源后才繼續 wait=False，立即返回，并不會等待池內的任務執行完畢 但不管wait參數為何值，整個程序都會等到所有任務執行完畢 submit和map必須在shutdown之前 4. result(timeout=None) 取得結果 5. add_done_callback(fn) 回調函數 6. done() 判斷某一個線程是否完成 7. cancle() 取消某個任務 ## 進程池/線程池用法舉例 線程池和進程池的用法完全一樣 ### 進程池案例 ~~~ from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ProcessPoolExecutor import os,time,random def task(n): print('%s is runing' %os.getpid()) time.sleep(random.randint(1,3)) return n**2 if __name__ == '__main__': executor=ProcessPoolExecutor(max_workers=3) futures=[] for i in range(6): future=executor.submit(task,i) futures.append(future) executor.shutdown(True) print('+++>') for future in futures: print(future.result()) ~~~ 執行結果: ``` 9312 is runing 9960 is runing 2720 is runing 2720 is runing 2720 is runing 9312 is runing +++> 0 1 4 9 16 25 ``` ### 線程池案例: ~~~ from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ProcessPoolExecutor import os,time,random def noah(n): print('線程:%s'%n) time.sleep(random.randint(1,5)) if __name__ == '__main__': thread=ThreadPoolExecutor(3) for i in range(6): thread.submit(noah,i) print('主.......') ~~~ 執行結果 ``` 線程0 線程1 線程2 主....... 線程3 線程4 線程5 ``` ## map方法和回調函數 ### map方法 將前面的for循環改為用map方法來實現 ~~~ from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ProcessPoolExecutor import os,time,random def noah(n): print('線程:%s'%n) time.sleep(random.randrange(0,2)) if __name__ == '__main__': thread=ThreadPoolExecutor(3) thread.map(noah,range(6)) thread.map(noah, range(6)) thread.shutdown() print('主.......') ~~~ 說明: 之前的Process模塊的map方法自帶jion方法,會等待子進程結束再運行后面的主進程,但是這個新模塊的map方法,不自帶shutdown功能,也就是說可以多次提交map后,用一個shutdown方法一起提交等待 ### 回調函數 可以為進程池或線程池內的每個進程或線程綁定一個函數，該函數在進程或線程的任務執行完畢后自動觸發，并接收任務的返回值當作參數，該函數稱為回調函數 ~~~ from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ProcessPoolExecutor import os,time,random def noah(n): # print('線程:%s'%n) time.sleep(random.randrange(0,2)) return n*n def call(m): print(m.result()) #拿到的是一個對象obj，需要用obj.result()拿到結果 if __name__ == '__main__': thread=ThreadPoolExecutor(3) for i in range(6): thread.submit(noah,i).add_done_callback(call) print('主.......') ~~~ ## **對比總結:** * **multiprocessing.Pool模塊** apply_async異步提交任務 得join，close之后才能維護主進程與池之間的同步 map自帶join，close效果 獲取執行結果，get 回調函數 指定callback參數，由主進程執行 * **concurrent.futures模塊** submit 異步提交任務 shutdown 同步 map是不自帶shutdown 獲取執行結果 result 回調函數 直接調用add_done_callback方法，由子線程/子進程執行 * **concurrent.futures優勢** 用concurrent.futures，可以輕松的在進程和線程之間切換 并發程序，線程池進程池都要用，只需要導入一個模塊 concurrent.futures是一個新的模塊，統一了線程池和進程池的使用方式，對一些操作進行了更合理的規劃 * 線程池一般的線程個數 ：**CPU個數*5** * 進程池一般的進程個數：**CPU個數+1**