1.為什么用執行緒池
1.啟動一個新執行緒的消耗較高且涉及與作業系統的互動,尤其是程式中需要創建大量生存期很短暫的執行緒,而使用執行緒池可以很好地提升性能
2.執行緒池則是創建指定執行緒數量等待執行事件,當該事件執行結束后該執行緒并不會死亡,而是回到執行緒池中變成空閑狀態等待執行下一個事件
3.當系統中包含有大量的并發執行緒時,會導致系統性能急劇下降甚至導致解釋器崩潰,而使用執行緒池可以有效地控制系統中并發執行緒的數量
2.執行緒池的使用步驟
1.threadpool 模塊實作執行緒池不推薦繼續使用,此處推薦是用 concurrent.futures 模塊中的 ThreadPoolExecutor 類實作執行緒池
2.呼叫 ThreadPoolExecutor 類的構造器創建一個執行緒池,定義一個普通函式作為執行緒任務
3.呼叫 ThreadPoolExecutor 物件的 submit() 方法來提交執行緒任務,呼叫 ThreadPoolExecutor 物件的 shutdown() 方法來關閉執行緒池
3.語法概述
from concurrent.futures import ThreadPoolExecutor thread_pool = ThreadPoolExecutor(max_workers=5) # 創建指定行程數量行程池并回傳行程池物件 future = thread_pool.submit(fn, *args, **kwargs) # 將fn函式提交給執行緒池,并回傳一個 future 物件 引數: *args 代表傳給 fn 函式的引數 *kwargs 代表以關鍵字引數的形式為 fn 函式傳入引數 # 該函式類似于內建函式 map(func, *iterables) 只是該函式將會啟動多個執行緒,以異步方式立即對 iterables 執行 map 處理 thread_pool.map(func, *iterables, timeout=None, chunksize=1) # 提交多個任務給池中,等效for + submit thread_pool.shutdown(wait=True) # 等待池內所有任務執行完畢后關閉執行緒池 future.cancel() # 取消該future代表的執行緒任務,如果該任務正在執行不可取消則該方法回傳False,否則程式會取消該任務并回傳True future.cancelled() # 回傳future代表的執行緒任務是否被成功取消 future.running() # 如果該future代表的執行緒任務正在執行不可被取消,該方法回傳True future.done() # 如果該funture代表的執行緒任務被成功取消或執行完成,則該方法回傳True future.result(timeout=None) # 獲取該future代表的執行緒任務最后回傳的結果,如果future代表的執行緒任務還未完成該方法將會阻塞當前執行緒 引數: timeout 指定最多阻塞等待多少秒 future.exception(timeout=None) # 獲取該future代表的執行緒任務所引發的例外,如果該任務成功完成沒有例外則該方法回傳None future.add_done_callback(fn) # 為該future代表的執行緒任務注冊一個回呼函式,當該任務成功完成時程式會自動觸發該fn函式
4.執行緒池的基本使用
import time from concurrent.futures import ThreadPoolExecutor def func(num): time.sleep(1) print("num is %s" % num) return num * num def main(): ret_list = list() executor = ThreadPoolExecutor(max_workers=5) # 指定執行緒池中的執行緒數量 for i in range(5): future = executor.submit(func, i) # 異步提交任務,回傳一個未來物件future ret_list.append(future) executor.shutdown(True) # 等待池內所有任務執行完畢回收完資源后才繼續 for ret in ret_list: ret = ret.result() # 獲取該future代表的執行緒任務最后回傳的結果 print(ret) if __name__ == "__main__": main() """執行結果 num is 1 num is 0 num is 3 num is 2 num is 4 0 1 4 9 16 """
5.執行緒池的多任務提交
from concurrent.futures import ThreadPoolExecutor import time def func(num): sum = 0 for i in range(num): sum += i ** 2 print(sum) t = ThreadPoolExecutor(20) start = time.time() t.map(func, range(1000)) # 提交多個任務給池中,等效于 for + submit t.shutdown() print(time.time() - start)
6.執行緒池的回傳值
from concurrent.futures import ThreadPoolExecutor import time def func(num): sum = 0 # time.sleep(5) # print(num) # 異步的效果 for i in range(num): sum += i ** 2 return sum t = ThreadPoolExecutor(20) # 下列代碼是用map的方式提交多個任務,對應拿結果的方法是__next__()回傳的是一個生成器物件 res = t.map(func, range(1000)) t.shutdown() print(res.__next__()) print(res.__next__()) print(res.__next__()) print(res.__next__()) # 下列代碼是用for + submit提交多個任務的方式,對應拿結果的方法是result t = ThreadPoolExecutor(20) res_l = [] for i in range(1000): re = t.submit(func, i) res_l.append(re) t.shutdown() [print(i.result()) for i in res_l] # 在Pool行程池中拿結果,是用get方法,在ThreadPoolExecutor里邊拿結果是用result方法
7.執行緒池中子執行緒呼叫回呼函式
from threading import current_thread from concurrent.futures import ThreadPoolExecutor # 在執行緒池中,回呼函式是子執行緒呼叫的,和父執行緒沒有關系 from concurrent.futures import ProcessPoolExecutor # 不管是ProcessPoolExecutor的行程池 還是Pool的行程池,回呼函式都是父行程呼叫的,和子行程沒有關系 import os def func(num): sum = 0 for i in range(num): sum += i ** 2 print('這是在子執行緒中', current_thread()) # current_thread()查看執行緒標識,類似于行程中的getpid() return sum def call_back_fun(res): # print(res.result(), os.getpid()) print('這是在回呼函式中', res.result(), current_thread()) # print(os.getpid()) if __name__ == '__main__': print(os.getpid()) t = ThreadPoolExecutor(20) # 執行緒池 # t = ProcessPoolExecutor(20) # 行程池 for i in range(10): t.submit(func, i).add_done_callback(call_back_fun) t.shutdown() print('這是在主執行緒中', current_thread())
8.行程池和執行緒池效率對比
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor from multiprocessing import Pool # concurrent.futures 這個模塊是異步呼叫的機制 # concurrent.futures 提交任務都是用submit # for + submit 多個任務的提交 # shutdown 是等效于Pool中的close+join,是指不允許再繼續向池中增加任務,然后讓父行程(執行緒)等待池中所有行程執行完所有任務 # from multiprocessing import Pool.apply / apply_async import time def func(num): sum = 0 for i in range(num): for j in range(i): for x in range(j): sum += x ** 2 # print(sum) if __name__ == '__main__': # pool的行程池的效率演示 p = Pool(5) start = time.time() for i in range(100): p.apply_async(func, args=(i,)) p.close() p.join() print('Pool行程池的效率時間是%s' % (time.time() - start)) # 0.51 # 多行程的效率演示 tp = ProcessPoolExecutor(5) start = time.time() for i in range(100): tp.submit(func, i) tp.shutdown() # 等效于行程池中的 close + join print('ProcessPoolExecutor行程池的消耗時間為%s' % (time.time() - start)) # 0.49 # 多執行緒的效率 tp = ThreadPoolExecutor(20) start = time.time() for i in range(100): tp.submit(func, i) tp.shutdown() # 等效于 行程池中的 close + join print('ThreadPoolExecutor執行緒池的消耗時間為%s' % (time.time() - start)) # 1.40 # 結果: 針對計算密集的程式來說 # 不管是Pool的行程池還是ProcessPoolExecutor()的行程池,執行效率相當 # ThreadPoolExecutor 的效率要差很多 # 所以當計算密集時,使用多行程
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/24497.html
標籤：Python
上一篇：并發編程（執行緒）——前言理論，開啟執行緒的兩種方式，執行緒物件join方法，同一個行程下的多個執行緒資料共享，執行緒物件及其他方法，守護執行緒，執行緒互斥鎖，GIL全域解釋器鎖理論