python中ThreadPoolExecutor線程池和ProcessPoolExecutor進程池
1、ThreadPoolExecutor多線程
<1>為什麼需要線程池呢?
- 對於io密集型,提高執行的效率。
- 線程的創建是需要消耗系統資源的。
所以線程池的思想就是:每個線程各自分配一個任務,剩下的任務排隊等待,當某個線程完成瞭任務的時候,排隊任務就可以安排給這個線程繼續執行。
<2>標準庫concurrent.futures模塊
它提供瞭ThreadPoolExecutor和ProcessPoolExecutor兩個類,
分別實現瞭對threading模塊和multiprocessing模塊的進一步抽象。
不僅可以幫我們自動調度線程,還可以做到:
- 主線程可以獲取某一個線程(或者任務)的狀態,以及返回值
- 當一個線程完成的時候,主線程能夠立即知道
- 讓多線程和多進程的編碼接口一致
<3>簡單使用
# -*-coding:utf-8 -*- from concurrent.futures import ThreadPoolExecutor import time # 參數times用來模擬網絡請求時間 def get_html(times): print("get page {}s finished".format(times)) return times # 創建線程池 # 設置線程池中最多能同時運行的線程數目,其他等待 executor = ThreadPoolExecutor(max_workers=2) # 通過submit函數提交執行的函數到線程池中,submit函數立即返回,不阻塞 # task1和task2是任務句柄 task1 = executor.submit( get_html, (2) ) task2 = executor.submit( get_html, (3) ) # done()方法用於判斷某個任務是否完成,bool型,完成返回True,沒有完成返回False print( task1.done() ) # cancel()方法用於取消某個任務,該任務沒有放到線程池中才能被取消,如果已經放進線程池子中,則不能被取消 # bool型,成功取消瞭返回True,沒有取消返回False print( task2.cancel() ) # result()方法可以獲取task的執行結果,前提是get_html()函數有返回值 print( task1.result() ) print( task2.result() ) # 結果: # get page 3s finished # get page 2s finished # True # False # 2 # 3
ThreadPoolExecutor類在構造實例的時候,傳入max_workers參數來設置線程池中最多能同時運行的線程數目
使用submit()函數來提交線程需要執行任務(函數名和參數)到線程池中,並返回該任務的句柄,
註意:submit()不是阻塞的,而是立即返回。
通過submit()函數返回的任務句柄,能夠使用done()方法判斷該任務是否結束,使用cancel()方法來取消,使用result()方法可以獲取任務的返回值,查看內部代碼,發現該方法是阻塞的
<4>as_completed(一次性獲取所有的結果)
上面雖然提供瞭判斷任務是否結束的方法,但是不能在主線程中一直判斷,有時候我們是得知某個任務結束瞭,就去獲取結果,而不是一直判斷每個任務有沒有結束。這時候就可以使用as_completed方法一次取出所有任務的結果。
# -*-coding:utf-8 -*- from concurrent.futures import ThreadPoolExecutor, as_completed import time # 參數times用來模擬網絡請求時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times # 創建線程池子 # 設置最多2個線程運行,其他等待 executor = ThreadPoolExecutor(max_workers=2) urls = [3,2,4] # 一次性把所有的任務都放進線程池,得到一個句柄,但是最多隻能同時執行2個任務 all_task = [ executor.submit(get_html,(each_url)) for each_url in urls ] for future in as_completed( all_task ): data = future.result() print("in main:get page {}s success".format(data)) # 結果 # get page 2s finished # in main:get page 2s success # get page 3s finished # in main:get page 3s success # get page 4s finished # in main:get page 4s success # 從結果可以看到,並不是先傳入哪個url,就先執行哪個url,沒有先後順序
<5>map()方法
除瞭上面的as_completed()方法,還可以使用execumap方法。但是有一點不同,使用map方法,不需提前使用submit方法,
map方法與python標準庫中的map含義相同,都是將序列中的每個元素都執行同一個函數。上面的代碼就是對urls列表中的每個元素都執行get_html()函數,並分配各線程池。可以看到執行結果與上面的as_completed方法的結果不同,輸出順序和urls列表的順序相同,就算2s的任務先執行完成,也會先打印出3s的任務先完成,再打印2s的任務完成
# -*-coding:utf-8 -*- from concurrent.futures import ThreadPoolExecutor,as_completed import time # 參數times用來模擬網絡請求時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times # 創建線程池子 # 設置最多2個線程運行,其他等待 executor = ThreadPoolExecutor(max_workers=2) urls = [3,2,4] for result in executor.map(get_html, urls): print("in main:get page {}s success".format(result))
結果:
get page 2s finished
get page 3s finished
in main:get page 3s success
in main:get page 2s success
get page 4s finished
in main:get page 4s success
<6>wait()方法
wait方法可以讓主線程阻塞,直到滿足設定的要求。wait方法接收3個參數,等待的任務序列、超時時間以及等待條件。
等待條件return_when默認為ALL_COMPLETED,表明要等待所有的任務都借宿。可以看到運行結果中,確實是所有任務都完成瞭,主線程才打印出main,等待條件還可以設置為FIRST_COMPLETED,表示第一個任務完成就停止等待。
超時時間參數可以不設置:
wait()方法和as_completed(), map()沒有關系。不管你是用as_completed(),還是用map()方法,你都可以在執行主線程之前使用wait()。
as_completed()和map()是二選一的。
# -*-coding:utf-8 -*- from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED,FIRST_COMPLETED import time # 參數times用來模擬網絡請求時間 def get_html(times): time.sleep(times) print("get page {}s finished".format(times)) return times # 創建線程池子 # 設置最多2個線程運行,其他等待 executor = ThreadPoolExecutor(max_workers=2) urls = [3,2,4] all_task = [executor.submit(get_html,(url)) for url in urls] wait(all_task,return_when=ALL_COMPLETED) print("main") # 結果 # get page 2s finished # get page 3s finished # get page 4s finished # main
2、ProcessPoolExecutor多進程
<1>同步調用方式: 調用,然後等返回值,能解耦,但是速度慢
import datetime from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor from threading import current_thread import time, random, os import requests def task(name): print('%s %s is running'%(name,os.getpid())) #print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) if __name__ == '__main__': p = ProcessPoolExecutor(4) # 設置 for i in range(10): # 同步調用方式,不僅要調用,還要等返回值 obj = p.submit(task, "進程pid:") # 傳參方式(任務名,參數),參數使用位置或者關鍵字參數 res = obj.result() p.shutdown(wait=True) # 關閉進程池的入口,等待池內任務運行結束 print("主") ################ ################ # 另一個同步調用的demo def get(url): print('%s GET %s' % (os.getpid(),url)) time.sleep(3) response = requests.get(url) if response.status_code == 200: res = response.text else: res = "下載失敗" return res # 有返回值 def parse(res): time.sleep(1) print("%s 解析結果為%s" %(os.getpid(),len(res))) if __name__ == "__main__": urls = [ 'https://www.baidu.com', 'https://www.sina.com.cn', 'https://www.tmall.com', 'https://www.jd.com', 'https://www.python.org', 'https://www.openstack.org', 'https://www.baidu.com', 'https://www.baidu.com', 'https://www.baidu.com', ] p=ProcessPoolExecutor(9) l=[] start = time.time() for url in urls: future = p.submit(get,url) # 需要等結果,所以是同步調用 l.append(future) # 關閉進程池,等所有的進程執行完畢 p.shutdown(wait=True) for future in l: parse(future.result()) print('完成時間:',time.time()-start) #完成時間: 13.209137678146362
<2>異步調用方式:隻調用,不等返回值,可能存在耦合,但是速度快
def task(name): print("%s %s is running" %(name,os.getpid())) time.sleep(random.randint(1,3)) if __name__ == '__main__': p = ProcessPoolExecutor(4) # 設置進程池內進程 for i in range(10): # 異步調用方式,隻調用,不等返回值 p.submit(task,'進程pid:') # 傳參方式(任務名,參數),參數使用位置參數或者關鍵字參數 p.shutdown(wait=True) # 關閉進程池的入口,等待池內任務運行結束 print('主') ################## ################## # 另一個異步調用的demo def get(url): print('%s GET %s' % (os.getpid(),url)) time.sleep(3) reponse = requests.get(url) if reponse.status_code == 200: res = reponse.text else: res = "下載失敗" parse(res) # 沒有返回值 def parse(res): time.sleep(1) print('%s 解析結果為%s' %(os.getpid(),len(res))) if __name__ == '__main__': urls = [ 'https://www.baidu.com', 'https://www.sina.com.cn', 'https://www.tmall.com', 'https://www.jd.com', 'https://www.python.org', 'https://www.openstack.org', 'https://www.baidu.com', 'https://www.baidu.com', 'https://www.baidu.com', ] p = ProcessPoolExecutor(9) start = time.time() for url in urls: future = p.submit(get,url) p.shutdown(wait=True) print("完成時間",time.time()-start)# 完成時間 6.293345212936401
<3>怎麼使用異步調用方式,但同時避免耦合的問題?
(1)進程池:異步 + 回調函數,,cpu密集型,同時執行,每個進程有不同的解釋器和內存空間,互不幹擾
def get(url): print('%s GET %s' % (os.getpid(), url)) time.sleep(3) response = requests.get(url) if response.status_code == 200: res = response.text else: res = '下載失敗' return res def parse(future): time.sleep(1) # 傳入的是個對象,獲取返回值 需要進行result操作 res = future.result() print("res",) print('%s 解析結果為%s' % (os.getpid(), len(res))) if __name__ == '__main__': urls = [ 'https://www.baidu.com', 'https://www.sina.com.cn', 'https://www.tmall.com', 'https://www.jd.com', 'https://www.python.org', 'https://www.openstack.org', 'https://www.baidu.com', 'https://www.baidu.com', 'https://www.baidu.com', ] p = ProcessPoolExecutor(9) start = time.time() for url in urls: future = p.submit(get,url) #模塊內的回調函數方法,parse會使用future對象的返回值,對象返回值是執行任務的返回值 #回調應該是相當於parse(future) future.add_done_callback(parse) p.shutdown(wait=True) print("完成時間",time.time()-start)#完成時間 33.79998469352722
(2)線程池:異步 + 回調函數,IO密集型主要使用方式,線程池:執行操作為誰有空誰執行
def get(url): print("%s GET %s" %(current_thread().name,url)) time.sleep(3) reponse = requests.get(url) if reponse.status_code == 200: res = reponse.text else: res = "下載失敗" return res def parse(future): time.sleep(1) res = future.result() print("%s 解析結果為%s" %(current_thread().name,len(res))) if __name__ == '__main__': urls = [ 'https://www.baidu.com', 'https://www.sina.com.cn', 'https://www.tmall.com', 'https://www.jd.com', 'https://www.python.org', 'https://www.openstack.org', 'https://www.baidu.com', 'https://www.baidu.com', 'https://www.baidu.com', ] p = ThreadPoolExecutor(4) start = time.time() for url in urls: future = p.submit(get,url) future.add_done_callback(parse) p.shutdown(wait=True) print("主",current_thread().name) print("完成時間",time.time()-start)#完成時間 32.52604126930237
3、總結
- 1、線程不是越多越好,會涉及cpu上下文的切換(會把上一次的記錄保存)。
- 2、進程比線程消耗資源,進程相當於一個工廠,工廠裡有很多人,裡面的人共同享受著福利資源,,一個進程裡默認隻有一個主線程,比如:開啟程序是進程,裡面執行的是線程,線程隻是一個進程創建多個人同時去工作。
- 3、線程裡有GIL全局解鎖器:不允許cpu調度
- 4、計算密度型適用於多進程
- 5、線程:線程是計算機中工作的最小單元
- 6、進程:默認有主線程 (幫工作)可以多線程共存
- 7、協程:一個線程,一個進程做多個任務,使用進程中一個線程去做多個任務,微線程
- 8、GIL全局解釋器鎖:保證同一時刻隻有一個線程被cpu調度
到此這篇關於python中ThreadPoolExecutor線程池和ProcessPoolExecutor進程池的文章就介紹到這瞭,更多相關python ThreadPoolExecutor 內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- python 多線程實現多任務的方法示例
- Python線程池thread pool創建使用及實例代碼分享
- Python 線程池模塊之多線程操作代碼
- python concurrent.futures模塊的使用測試
- python協程與 asyncio 庫詳情