前言

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

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

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

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

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

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

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

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

Futures簡介

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

Futures的基本用法

要使用Futures，首先需要導(dǎo)入concurrent.futures模塊：

from concurrent.futures 
import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

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

下面是一個使用ThreadPoolExecutor的示例，計(jì)算斐波那契數(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()方法將任務(wù)fibonacci(i)提交給線程池，返回一個Future對象。通過遍歷所有的Future對象，并使用as_completed()函數(shù)等待并獲取任務(wù)的結(jié)果，我們可以得到每個任務(wù)的計(jì)算結(jié)果。

這里我們創(chuàng)建了一個線程池，總共有 2個線程可以分配使用。雖然線程的數(shù)量可以自己定義，但是線程數(shù)并不是越多越好，因?yàn)榫€程的創(chuàng)建、維護(hù)和刪除也會有一定的開銷。所以如果你設(shè)置的很大，反而可能會導(dǎo)致速度變慢。我們往往需要根據(jù)實(shí)際的需求做一些測試，來尋找最優(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)點(diǎn)是簡單明了，但是明顯效率低下

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

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

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

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

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
?
 ? ?# 阻塞主線程，等待任務(wù)的完成
 ? ?futures.wait([future])

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

控制并發(fā)度

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

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

下面是一個控制并發(fā)度的示例，通過設(shè)置max_workers為2，限制了同時執(zhí)行的任務(wù)數(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í)行兩個任務(wù)，其余的任務(wù)會等待前面的任務(wù)完成后再執(zhí)行。

錯誤處理

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

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

下面是一個處理任務(wù)異常的示例：

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)

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

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

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

最后

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

再舉個race condition的例子

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

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

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)
?

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

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

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

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

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

最新評論