前言

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

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

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

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

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

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

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

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

Futures簡(jiǎn)介

Futures是Python標(biāo)準(zhǔn)庫中concurrent.futures模塊提供的一種并發(fā)編程概念。它允許我們?cè)谥骶€程中定義任務(wù)，并在后臺(tái)進(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

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

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

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

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

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

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

下面是一個(gè)使用回調(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()方法注冊(cè)了一個(gè)回調(diào)函數(shù)。當(dāng)任務(wù)完成時(shí)，回調(diào)函數(shù)會(huì)被自動(dòng)調(diào)用，并傳遞Future對(duì)象作為參數(shù)。我們可以在回調(diào)函數(shù)中使用future.result()獲取任務(wù)的計(jì)算結(jié)果。

控制并發(fā)度

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

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

下面是一個(gè)控制并發(fā)度的示例，通過設(shè)置max_workers為2，限制了同時(shí)執(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)建了一個(gè)最大并發(fā)度為2的線程池。這意味著最多同時(shí)執(zhí)行兩個(gè)任務(wù)，其余的任務(wù)會(huì)等待前面的任務(wù)完成后再執(zhí)行。

錯(cuò)誤處理

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

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

下面是一個(gè)處理任務(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é)果時(shí)，使用try-except語句捕獲異常，并進(jìn)行處理。

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

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

最后

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

再舉個(gè)race condition的例子

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

假設(shè)有一個(gè)共享的計(jì)數(shù)器變量 count，初始值為0?，F(xiàn)在有兩個(gè)線程同時(shí)對(duì) 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，即兩個(gè)線程各加了1000000次，但是實(shí)際執(zhí)行結(jié)果并不是這樣。這就是Race condition問題。

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

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

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

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

最新評(píng)論