Python線程池的正確使用

1、為什么要使用線程池呢？

因為線程執行完任務之后就會被系統銷毀，下次再執行任務的時候再進行創建。這種方式在邏輯上沒有啥問題。但是系統啟動一個新線程的成本是比較高，因為其中涉及與操作系統的交互，操作系統需要給新線程分配資源。打個比方吧！就像軟件公司招聘員工干活一樣。當有活干時，就招聘一個外包人員干活。當活干完之后就把這個人員辭退掉。你說在這過程中所耗費的時間成本和溝通成本是不是很大。那么公司一般的做法是：當項目立項時就確定需要幾名開發人員，然后將這些人員配齊。然后這些人員就常駐在項目組，有活就干，沒活就摸魚。線程池也是同樣的道理。線程池可以定義最大線程數，這些線程有任務就執行任務，沒任務就進入線程池中歇著。

2、線程池怎么用呢？

線程池的基類是concurrent.futures模塊中的Executor類，而Executor類提供了兩個子類，即ThreadPoolExecutor類和ProcessPoolExecutor類。其中ThreadPoolExecutor用于創建線程池，而ProcessPoolExecutor用于創建進程池。本文將重點介紹ThreadPoolExecutor類的使用。首先，讓我們來看看ThreadPoolExecutor類的構造函數。這里使用的Python版本是:3.6.7。

      def __init__(self, max_workers=None, thread_name_prefix=''):
        """Initializes a new ThreadPoolExecutor instance.

        Args:
            max_workers: The maximum number of threads that can be used to
                execute the given calls.
            thread_name_prefix: An optional name prefix to give our threads.
        """
        if max_workers is None:
            # Use this number because ThreadPoolExecutor is often
            # used to overlap I/O instead of CPU work.
            max_workers = (os.cpu_count() or 1) * 5
        if max_workers = 0:
            raise ValueError("max_workers must be greater than 0")

        self._max_workers = max_workers
        self._work_queue = queue.Queue()
        self._threads = set()
        self._shutdown = False
        self._shutdown_lock = threading.Lock()
        self._thread_name_prefix = (thread_name_prefix or
                                    ("ThreadPoolExecutor-%d" % self._counter()))

他的構造函數只有兩個參數：一個是max_workers參數，用于指定線程池的最大線程數，如果不指定的話則默認是CPU核數的5倍。另一個參數是thread_name_prefix，它用來指定線程池中線程的名稱前綴。其他參數：

• _shutdown初始值值為False,默認情況下線程池不銷毀，即線程池的生命周期跟項目的生命周期一致。
• self._work_queue = queue.Queue()生成緩沖隊列。
• _threads沒有任務被提交時，線程的數量設置為0。
• _shutdown_lock 指定線程池的鎖是Lock鎖。
• 說完了線程池的創建之后，接著來看看線程池中比較常用的幾個方法吧。
• submit(self, fn, *args, **kwargs)：

• 該方法用提交任務，即將fn函數提交給線程池，*args代表傳給fn函數的參數，**kwargs代表以關鍵字參數的形式為fn函數傳入參數。
• shutdown(self, wait=True)：
• 關閉線程池
• map(func, *iterables, timeout=None, chunksize=1)：
• 該函數類似于全局函數map(func,*iterables),只是該函數將會啟動多個線程，以異步方式立即對iterables執行map處理。

程序將task函數通過submit方法提交給線程池之后，線程池會返回一個Future對象，該對象的作用主要是用于獲取線程任務函數的返回值。Future提供了如下幾個方法。

• cancel()：取消該Future代表的線程任務。如果該任務正在執行，不可取消，則該方法返回False;否則，程序會取消該任務，并返回True。
• result(timeout=None):獲取該 Future 代表的線程任務最后返回的結果。如果 Future 代表的線程任務還未完成，該方法將會阻塞當前線程，其中 timeout 參數指定最多阻塞多少秒。
• add_done_callback(fn)：為該 Future 代表的線程任務注冊一個“回調函數”，當該任務成功完成時，程序會自動觸發該 fn 函數。
• done():如果該Future代表的線程任務被成功取消或執行完成，則該方法返回True。

來個簡單的例子：

該例中創建了一個最大線程數是2的線程池來執行async_add函數。

from concurrent.futures import ThreadPoolExecutor
import threading
import time


def async_add(max):
    sum = 0
    for i in range(max):
        sum = sum + i
    time.sleep(1)
    print(threading.current_thread().name + "執行求和操作求得的和是=" + str(sum))
    return sum

# 創建兩個線程
pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix='測試線程')
# 向線程池提交一個task,20作為async_add()函數的參數
future1 = pool.submit(async_add, 20)
# 向線程池再提交一個task
future2 = pool.submit(async_add, 50)
# 判斷future1代表的任務是否執行完
time.sleep(2)
print(future1.done())
print(future2.done())
# 查看future1代表的任務返回的結果
print('線程一的執行結果是=' + str(future1.result()))
# 查看future2代表的任務的返回結果
print('線程二的執行結果是=' + str(future2.result()))
print("----" + threading.current_thread().name + "----主線程執行結束-----")

運行結果是：

測試線程_0執行求和操作求得的和是=190
測試線程_1執行求和操作求得的和是=1225
True
True
線程一的執行結果是=190
線程二的執行結果是=1225
----MainThread----主線程執行結束-----

本例中定義了一個最大線程數是2的線程池，并向線程池中提交了兩個任務，其中async_add函數就是要執行的任務。在async_add函數中添加 time.sleep(1) 休眠一秒是為了驗證done()方法返回的結果。最后才打印主線程執行結束表明result()方法是阻塞的。如果將result()屏蔽掉。
改成如下形式：

# 創建兩個線程
pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix='測試線程')
# 向線程池提交一個task,20作為async_add()函數的參數
future1 = pool.submit(async_add, 20)
# 向線程池再提交一個task
future2 = pool.submit(async_add, 50)
# 判斷future1代表的任務是否執行完
print(future1.done())
print(future2.done())
print("----" + threading.current_thread().name + "----主線程執行結束-----")

則運行結果是：

False
False
----MainThread----主線程執行結束-----
測試線程_0執行求和操作求得的和是=190
測試線程_1執行求和操作求得的和是=1225

3、如何非阻塞的獲取線程執行的結果

前面介紹的result()方法是通過阻塞的方式來獲取線程的運行結果的。那么如果通過非阻塞的方法來獲取線程任務最后的返回結果呢？這里就需要使用線程的回調函數來獲取線程的返回結果。

from concurrent.futures import ThreadPoolExecutor
import threading
import time


def async_add(max):
    sum = 0
    for i in range(max):
        sum = sum + i
    time.sleep(1)
    print(threading.current_thread().name + "執行求和操作求得的和是=" + str(sum))
    return sum


with ThreadPoolExecutor(max_workers=2) as pool:
    # 向線程池提交一個task
    future1 = pool.submit(async_add, 20)
    future2 = pool.submit(async_add, 50)


    # 定義獲取結果的函數
    def get_result(future):
        print(threading.current_thread().name + '運行結果：' + str(future.result()))


    # 查看future1代表的任務返回的結果
    future1.add_done_callback(get_result)
    # 查看future2代表的任務的返回結果
    future2.add_done_callback(get_result)
    print('------------主線程執行結束----')

運行結果是：

------------主線程執行結束----
ThreadPoolExecutor-0_1執行求和操作求得的和是=1225
ThreadPoolExecutor-0_1運行結果：1225
ThreadPoolExecutor-0_0執行求和操作求得的和是=190
ThreadPoolExecutor-0_0運行結果：190

從結果可以看出獲取線程執行結果的方法完全沒有阻塞到主線程的運行。這里通過add_done_callback函數向線程池中注冊了一個獲取線程執行結果的函數get_result。
由于線程池實現了上下文管理協議（Context Manage Protocol）,因此程序可以使用with語句來管理線程池，這樣即可避免手動關閉線程池。

4、線程池的運行策略

這里有必要介紹一下線程池的執行策略，也就是說當線程池中的任務數大于線程池的最大線程數時，線程池該如何處理這些任務呢？處理不了的任務是直接丟棄還是慢慢處理呢？再回答這個問題之前，讓我們來看下下面這個例子：這里定義了一個最大線程數是4個線程池，然后向線程池中提交了100個task任務。

def async_add(max):
    sum = 0
    for i in range(max):
        sum = sum + i
    time.sleep(1)
    print(threading.current_thread().name + "執行求和操作求得的和是=" + str(sum))
    return sum


with ThreadPoolExecutor(max_workers=4) as pool:
    for i in range(100):
        pool.submit(async_add, i)
    print('------------主線程執行結束----')

運行結果是：

------------主線程執行結束----
ThreadPoolExecutor-0_1執行求和操作求得的和是=0
ThreadPoolExecutor-0_0執行求和操作求得的和是=0
ThreadPoolExecutor-0_3執行求和操作求得的和是=3
ThreadPoolExecutor-0_2執行求和操作求得的和是=1
...省略部分結果.....
ThreadPoolExecutor-0_1執行求和操作求得的和是=4656
ThreadPoolExecutor-0_2執行求和操作求得的和是=4753
ThreadPoolExecutor-0_0執行求和操作求得的和是=4560
ThreadPoolExecutor-0_3執行求和操作求得的和是=4851

從運行結果可以看出：一直都是相同的線程來執行這些任務，并且所有的任務都沒有被丟棄。并且任務按照先來后到的順序來執行。這里就需要說到線程池默認的緩沖隊列了。self._work_queue = queue.Queue() 該語句會創建一個大小無限制的緩沖隊列。該隊列是一個 FIFO（先進先出）的常規隊列。所以當任務數超過最大線程數時，任務會暫時放在緩沖隊列queue中。當線程空閑之后會從緩沖隊列中取出任務來執行。
該隊列有個參數maxsize可以限制隊列的大小。如果隊列的大小達到隊列的上限，就會加鎖，再次加入元素時，就會被阻塞，直到隊列中的元素被消費。如果將maxsize的設置為0或者負數時，則該隊列的大小就是無限制的。

到此這篇關于Python線程池的正確使用方法的文章就介紹到這了,更多相關Python線程池的正確使用內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！