前言

Python是一門流行且強大的編程語言，具備靈活的異步編程能力。在并發(fā)編程中，F(xiàn)utures模塊是Python提供的一個強大工具，它簡化了異步編程的復雜性，使得編寫并發(fā)代碼變得更加直觀和易于閱讀。本文將介紹Futures的基本概念和用法，并通過實例來引導讀者深入理解。

區(qū)別并發(fā)與并行

并發(fā)指的是同時處理多個任務的能力。在并發(fā)執(zhí)行中，任務按照交替、間斷的方式進行，看起來好像是同時執(zhí)行。通過在任務之間快速切換，實現(xiàn)了看似同時進行的效果。并發(fā)適用于單核處理器系統(tǒng)，其中只有一個物理處理單元。

并行指的是真正同時執(zhí)行多個任務的能力。在并行執(zhí)行中，任務可以在多個物理處理單元（例如多核處理器、多個計算機節(jié)點等）上同時進行，每個任務都有自己的處理資源。并行適用于多核處理器系統(tǒng)，可以充分利用多個處理單元提高任務的執(zhí)行效率。

簡而言之， 并發(fā)指的是任務的交替執(zhí)行，通過任務之間的快速切換來實現(xiàn)并行的假象；并行則是真正的同時執(zhí)行多個任務。

以下是并發(fā)和并行的一些關鍵區(qū)別：

• 單元數(shù)量：并發(fā)適用于單個處理單元（如單核CPU），并行適用于多個處理單元（如多核CPU）。
• 實際執(zhí)行：并發(fā)是任務交替執(zhí)行，看起來像是同時進行；并行是真正同時執(zhí)行多個任務。
• 物理資源：并發(fā)共享物理資源，任務之間快速切換；并行每個任務都有自己的物理資源。
• 提高性能：并行可以通過同時執(zhí)行多個任務來提高整體性能，而并發(fā)主要用于提高響應性和減少等待時間。

并發(fā)，是指遇到I/O阻塞時（一般是網(wǎng)絡I/O或磁盤I/O），通過多個線程之間切換執(zhí)行多個任務（多線程）或單線程內多個任務之間切換執(zhí)行的方式來最大化利用CPU時間，但同一時刻，只允許有一個線程或任務執(zhí)行。適合I/O阻塞頻繁的業(yè)務場景。 并行，是指多個進程完全同步同時的執(zhí)行。適合CPU密集型業(yè)務場景。

Futures簡介

Futures是Python標準庫中concurrent.futures模塊提供的一種并發(fā)編程概念。它允許我們在主線程中定義任務，并在后臺進行并發(fā)執(zhí)行。通過使用Futures，我們可以異步地執(zhí)行任務、獲取任務的結果，以及處理異常情況，而無需顯式地管理線程或進程。

Futures的基本用法

要使用Futures，首先需要導入concurrent.futures模塊：

from concurrent.futures 
import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

接下來，我們需要創(chuàng)建一個線程池或進程池對象，以便執(zhí)行我們定義的任務。ThreadPoolExecutor用于開啟線程并執(zhí)行任務，ProcessPoolExecutor則用于開啟進程。

下面是一個使用ThreadPoolExecutor的示例，計算斐波那契數(shù)列的值：

def fibonacci(n):
 ? ?if n <= 1:
 ? ? ? ?return n
 ? ?else:
 ? ? ? ?return fibonacci(n-1) + fibonacci(n-2)
?
def main():
    start_time = time.perf_counter()
 ? ?with ThreadPoolExecutor(max_workers=5) as executor:
 ? ? ? ?futures = [executor.submit(fibonacci, i) for i in range(10)]
 ? ? ? ?results = [future.result() for future in as_completed(futures)]
?
 ? ?print(results)
 ? ?end_time = time.perf_counter()
?
 ? ?print(end_time - start_time) # 0.0010214539999999772

在上述示例中，我們使用executor.submit()方法將任務fibonacci(i)提交給線程池，返回一個Future對象。通過遍歷所有的Future對象，并使用as_completed()函數(shù)等待并獲取任務的結果，我們可以得到每個任務的計算結果。

這里我們創(chuàng)建了一個線程池，總共有 2個線程可以分配使用。雖然線程的數(shù)量可以自己定義，但是線程數(shù)并不是越多越好，因為線程的創(chuàng)建、維護和刪除也會有一定的開銷。所以如果你設置的很大，反而可能會導致速度變慢。我們往往需要根據(jù)實際的需求做一些測試，來尋找最優(yōu)的線程數(shù)量。

多線程與單線程的比較

import time
?
def fibonacci(n):
 ? ?if n <= 1:
 ? ? ? ?return n
 ? ?else:
 ? ? ? ?return fibonacci(n-1) + fibonacci(n-2)
?
def main():
 ? ?start_time = time.perf_counter()
 ? ?for i in range(10):
 ? ? ? ?fibonacci(i)
 ? ?end_time = time.perf_counter()
?
 ? ?print(end_time - start_time) ?# 5.414600000000491

這是單線程的執(zhí)行時間，單線程的優(yōu)點是簡單明了，但是明顯效率低下

異步執(zhí)行和獲取結果

Futures提供了異步執(zhí)行任務的能力。調用executor.submit()方法時，任務會立即開始執(zhí)行，并返回一個Future對象，代表該任務的未來結果。

要獲取任務的結果，可以使用Future對象的result()方法。它會阻塞主線程，直到任務完成并返回結果。另外，還可以使用add_done_callback()方法注冊一個回調函數(shù)，當任務完成時自動觸發(fā)。

下面是一個使用回調函數(shù)處理任務結果的示例：

def callback(future):
 ? ?result = future.result()
 ? ?print("Task completed with result:", result)
?
def main():
 ? ?with ThreadPoolExecutor() as executor:
 ? ? ? ?future = executor.submit(fibonacci, 10)
 ? ? ? ?future.add_done_callback(callback)
?
 ? ?# Do other work here
?
 ? ?# 阻塞主線程，等待任務的完成
 ? ?futures.wait([future])

在上述示例中，我們通過future.add_done_callback()方法注冊了一個回調函數(shù)。當任務完成時，回調函數(shù)會被自動調用，并傳遞Future對象作為參數(shù)。我們可以在回調函數(shù)中使用future.result()獲取任務的計算結果。

控制并發(fā)度

Futures允許我們通過控制并發(fā)度來管理并發(fā)執(zhí)行的任務。并發(fā)度指的是同時進行的任務數(shù)量。

通過傳遞max_workers參數(shù)給線程池或進程池對象，我們可以限制并發(fā)執(zhí)行的任務數(shù)量。如果未指定max_workers，則默認使用系統(tǒng)可用的核心數(shù)。

下面是一個控制并發(fā)度的示例，通過設置max_workers為2，限制了同時執(zhí)行的任務數(shù)量：

def main():
 ? ?with ThreadPoolExecutor(max_workers=2) as executor:
 ? ? ? ?futures = [executor.submit(fibonacci, i) for i in range(10)]
 ? ? ? ?results = [future.result() for future in as_completed(futures)]
?
 ? ?print(results)

在上述示例中，我們使用max_workers=2創(chuàng)建了一個最大并發(fā)度為2的線程池。這意味著最多同時執(zhí)行兩個任務，其余的任務會等待前面的任務完成后再執(zhí)行。

錯誤處理

當任務執(zhí)行過程中出現(xiàn)異常時，F(xiàn)utures提供了異常處理機制，使得我們可以捕獲和處理任務中的異常情況。

可以通過在任務函數(shù)內部拋出異常，或者使用future.set_exception()方法手動設置異常，來模擬任務的執(zhí)行失敗。

下面是一個處理任務異常的示例：

def task():
 ? ?raise Exception("Task execution failed")
?
def main():
 ? ?with ThreadPoolExecutor() as executor:
 ? ? ? ?future = executor.submit(task)
 ? ? ? ?try:
 ? ? ? ? ? ?result = future.result()
 ? ? ? ?except Exception as e:
 ? ? ? ? ? ?print("Task execution failed:", e)

在上述示例中，我們通過在任務函數(shù)task()中拋出異常模擬了任務執(zhí)行失敗的情況。在獲取任務結果時，使用try-except語句捕獲異常，并進行處理。

為啥多線程每次只有一個線程執(zhí)行

同一時刻，Python 主程序只允許有一個線程執(zhí)行，所以 Python 的并發(fā)，是通過多線程的切換完成的。你可能會疑惑這到底是為什么呢？這里我簡單提一下全局解釋器鎖的概念。事實上，Python 的解釋器并不是線程安全的，為了解決由此帶來的 race condition 等問題，Python 便引入了全局解釋器鎖，也就是同一時刻，只允許一個線程執(zhí)行。當然，在執(zhí)行 I/O 操作時，如果一個線程被 block 了，全局解釋器鎖便會被釋放，從而讓另一個線程能夠繼續(xù)執(zhí)行。

最后

Futures是Python并發(fā)編程的一個強大工具，它簡化了異步編程的復雜性，使得編寫并發(fā)代碼變得更加直觀和易于閱讀。本文介紹了Futures的基本用法，包括任務提交和執(zhí)行、獲取結果、控制并發(fā)度以及錯誤處理等方面。

再舉個race condition的例子

"Race condition"是并發(fā)編程中的一個常見問題，指的是多個進程或線程之間的執(zhí)行順序不確定，導致程序的行為變得不可預測。

假設有一個共享的計數(shù)器變量 count，初始值為0?，F(xiàn)在有兩個線程同時對 count 進行增加操作，代碼如下

import threading
?
count = 0
?
def increment():
 ? ?global count
 ? ?for _ in range(1000000):
 ? ? ? ?count += 1
?
thread1 = threading.Thread(target=increment)
thread2 = threading.Thread(target=increment)
?
thread1.start()
thread2.start()
?
thread1.join()
thread2.join()
?
print("Final count:", count)
?

預期結果應該是2000000，即兩個線程各加了1000000次，但是實際執(zhí)行結果并不是這樣。這就是Race condition問題。

在這個例子中，當兩個線程同時執(zhí)行 count += 1 操作時，它們會讀取當前的 count 值，然后分別進行加一操作，最后再寫回 count。然而，由于線程之間的切換和調度是不確定的，可能在某個線程讀取 count 后，還未來得及進行加一操作之前，另一個線程也已經(jīng)讀取了 count ，這樣導致了Race condition。

對于 threading，操作系統(tǒng)知道每個線程的所有信息，因此它會做主在適當?shù)臅r候做線程切換。很顯然，這樣的好處是代碼容易書寫，因為程序員不需要做任何切換操作的處理；但是切換線程的操作，容易出現(xiàn)在語句執(zhí)行過程中，這樣就容易出現(xiàn) race condition 的情況。

而對于 asyncio，主程序想要切換任務時，必須得到此任務可以被切換的通知，這樣一來也就可以避免剛剛提到的 race condition 的情況。

到此這篇關于Python并發(fā)編程之Futures模塊詳解的文章就介紹到這了,更多相關Python Futures內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論