PythonThreadPoolExecutor線程池問題怎麼解決

WBOY
發布: 2023-04-28 22:40:20
轉載
1606 人瀏覽過

    概念

    Python中已經有了threading模組,為什麼還需要執行緒池呢,線程池又是什麼東西呢?

    以爬蟲為例,需要控制同時爬取的線程數,例子中創建了20個線程,而同時只允許3個線程在運行,但是20個線程都需要創建和銷毀,線程的創建是需要消耗系統資源的,有沒有更好的方案呢?

    其實只需要三個執行緒就行了,每個執行緒各分配一個任務,剩下的任務排隊等待,當某個執行緒完成了任務的時候,排隊任務就可以安排給這個執行緒繼續執行。

    這就是線程池的想法(當然沒這麼簡單),但是自己寫線程池很難寫的比較完美,還需要考慮複雜情況下的線程同步,很容易發生死鎖。

    Python3.2開始,標準函式庫為我們提供了concurrent.futures模組,它提供了ThreadPoolExecutorProcessPoolExecutor 兩個類,實現了對threadingmultiprocessing的進一步抽象(這裡主要關注線程池),不僅可以幫我們自動調度線程,還可以做到:

    • 主執行緒可以取得某一個執行緒(或任務的)的狀態,以及回傳值。

    • 當一個執行緒完成的時候,主執行緒能夠立即知道。

    • 讓多執行緒和多進程的編碼介面一致。

    實例

    簡單使用

    from concurrent.futures import ThreadPoolExecutor
    import time
     
    # 参数times用来模拟网络请求的时间
    def get_html(times):
        time.sleep(times)
        print("get page {}s finished".format(times))
        return times
     
    executor = ThreadPoolExecutor(max_workers=2)
    # 通过submit函数提交执行的函数到线程池中,submit函数立即返回,不阻塞
    task1 = executor.submit(get_html, (3))
    task2 = executor.submit(get_html, (2))
    # done方法用于判定某个任务是否完成
    print(task1.done())
    # cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
    print(task2.cancel())
    time.sleep(4)
    print(task1.done())
    # result方法可以获取task的执行结果
    print(task1.result())
     
    # 执行结果
    # False  # 表明task1未执行完成
    # False  # 表明task2取消失败,因为已经放入了线程池中
    # get page 2s finished
    # get page 3s finished
    # True  # 由于在get page 3s finished之后才打印,所以此时task1必然完成了
    # 3     # 得到task1的任务返回值
    登入後複製

    ThreadPoolExecutor建構實例的時候,傳入max_workers參數來設定執行緒池中最多能同時執行的執行緒數目。

    使用submit函數來提交執行緒需要執行的任務(函數名稱和參數)到執行緒池中,並傳回該任務的句柄(類似檔案、畫圖),注意submit()不是阻塞的,而是立即返回。

    透過submit函數傳回的任務句柄,能夠使用done()方法判斷該任務是否結束。上面的例子可以看出,由於任務有2s的延時,在task1提交後立刻判斷,task1還未完成,而在延時4s之後判斷,task1就完成了。

    使用cancel()方法可以取消提交的任務,如果任務已經在執行緒池中運行了,就取消不了。這個例子中,線程池的大小設定為2,任務已經在運行了,所以取消失敗。如果改變線程池的大小為1,那麼先提交的是task1,task2還在排隊等候,這是時候就可以成功取消。

    使用result()方法可以取得任務的回傳值。查看內部程式碼,發現這個方法是阻塞的。

    as_completed

    上面雖然提供了判斷任務是否結束的方法,但是不能在主執行緒一直判斷啊。

    有時候我們是得知某個任務結束了,就去取得結果,而不是一直判斷每個任務有沒有結束。

    這是就可以使用as_completed方法一次取出所有任務的結果。

    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
     
    executor = ThreadPoolExecutor(max_workers=2)
    urls = [3, 2, 4] # 并不是真的url
    all_task = [executor.submit(get_html, (url)) for 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
    登入後複製

    as_completed()方法是一個生成器,在沒有任務完成的時候,會阻塞,在有某個任務完成的時候,會yield這個任務,就能執行for迴圈下面的語句,然後繼續阻塞住,再循環到所有的任務結束。

    從結果也可以看出,先完成的任務會先通知主執行緒

    map

    除了上面的as_completed方法,還可以使用executor.map方法,但是有一點不同。

    from concurrent.futures import ThreadPoolExecutor
    import time
     
    # 参数times用来模拟网络请求的时间
    def get_html(times):
        time.sleep(times)
        print("get page {}s finished".format(times))
        return times
     
    executor = ThreadPoolExecutor(max_workers=2)
    urls = [3, 2, 4] # 并不是真的url
     
    for data in executor.map(get_html, urls):
        print("in main: get page {}s success".format(data))
    # 执行结果
    # 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
    登入後複製

    使用map方法,無需事先使用submit方法,map方法與python標準函式庫中的map意義相同,都是將序列中的每個元素都執行同一個函數。

    上面的程式碼就是對urls的每個元素都執行get_html函數,並且分配各線程池。可以看到執行結果與上面的as_completed方法的結果不同,輸出順序和urls列表的順序相同,就算2s的任務先執行完成,也會先列印出3s的任務先完成,再列印2s的任務完成。

    wait

    wait方法可以讓主執行緒阻塞,直到滿足設定的要求。

    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
     
    executor = ThreadPoolExecutor(max_workers=2)
    urls = [3, 2, 4] # 并不是真的url
    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
    登入後複製

    wait方法接收3個參數,等待的任務序列、逾時時間、等待條件。

    等待條件return_when預設為ALL_COMPLETED,表示要等待所有的任務都結束。

    可以看到運行結果中,確實是所有任務都完成了,主執行緒才印出main

    等待條件也可以設定為FIRST_COMPLETED,表示第一個任務完成就停止等待。

    原始碼分析

    cocurrent.future模組中的future的意思是未來對象,可以把它理解為一個在未來完成的操作,這是非同步程式設計的基礎。

    在執行緒池submit()之後,回傳的就是這個future對象,回傳的時候任務並沒有完成,但會在將來完成。

    也可以稱為task的回傳容器,這個裡面會儲存task的結果和狀態。

    ThreadPoolExecutor內部是如何操作這個物件的呢?

    下面簡單介紹ThreadPoolExecutor的部分程式碼:

    1.init方法

    init方法中主要重要的就是任務隊列和線程集合,在其他方法中需要使用到。

    PythonThreadPoolExecutor線程池問題怎麼解決

    2.submit方法

    submit中有兩個重要的對象,_base.Future()_WorkItem()對象,_WorkItem()對象負責執行任務和對future對象進行設置,最後會將future對象返回,可以看到整個過程是立即返回的,沒有阻塞。

    PythonThreadPoolExecutor線程池問題怎麼解決

    3.adjust_thread_count方法

    這個方法的意義很好理解,主要是建立指定的執行緒數。但實作上有點難以理解,例如執行緒執行函數中的weakref.ref,涉及到了弱引用等概念,留待以後理解。

    PythonThreadPoolExecutor線程池問題怎麼解決

    4._WorkItem物件

    #_WorkItem物件的職責就是執行任務和設定結果。這裡面主要複雜的還是self.future.set_result(result)

    PythonThreadPoolExecutor線程池問題怎麼解決

    5.執行緒執行函數--_worker

    #這是執行緒池建立執行緒時指定的函數入口,主要是從佇列中依次取出task執行,但是函數的第一個參數還不是很明白。留待以後。

    PythonThreadPoolExecutor線程池問題怎麼解決

    以上是PythonThreadPoolExecutor線程池問題怎麼解決的詳細內容。更多資訊請關注PHP中文網其他相關文章!

    相關標籤:
    來源:yisu.com
    本網站聲明
    本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
    熱門教學
    更多>
    最新下載
    更多>
    網站特效
    網站源碼
    網站素材
    前端模板