Python 編程：多線程為 for 循環提速
for 循環是編程的一個基本方面，它允許我們迭代序列并高效地執行操作。然而，在處理耗時任務時，for 循環的順序性質可能成為瓶頸。一個解決方案是使用線程。學習：如何使用、何時使用以及何時不使用線程。像往常一樣，你可以在我的 GIT 倉庫中找到代碼示例。鏈接在頁腳。
讓我們從一個例子開始。我們將偽造并模擬一個耗時的任務。我們將使用一個 Python 腳本，該腳本通過 for 循環對數字列表進行處理，通過 square_number 函數將每個數字平方：
import time
# List of numbers to process
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Function to square a number
def square_number(number):
time.sleep(1) # Simulate a time-consuming task
return number * number
# Using a for loop to process each number
squared_numbers = []
start_time = time.time()
for number in numbers:
squared_numbers.append(square_number(number))
end_time = time.time()
print("Squared numbers:", squared_numbers)
print("Time taken:", end_time - start_time, "seconds")
# Time taken: 10.082990884780884 seconds這個腳本按順序處理列表中的每個數字，由于 square_number 函數中的 time.sleep(1) 調用，每個數字耗時 1 秒。總執行時間為 10.1 秒。
使用多線程優化
接下來，我們將使用多線程方法來優化這一點，以改善處理時間。為了使用多線程優化上述示例，我們可以使用 Python 的 concurrent.futures 模塊，它為異步執行可調用對象提供了一個高級接口。以下是如何修改腳本以使用多線程：
import time
from concurrent.futures import ThreadPoolExecutor
# List of numbers to process
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Function to square a number
def square_number(number):
time.sleep(1) # Simulate a time-consuming task
return number * number
# Using ThreadPoolExecutor for multithreading
squared_numbers = []
start_time = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
results = executor.map(square_number, numbers)
# Collect the results
squared_numbers = list(results)
end_time = time.time()
print("Squared numbers:", squared_numbers)
print("Time taken:", end_time - start_time, "seconds")
# Time taken: 2.0257720947265625 seconds在這個優化的腳本中，我們使用 ThreadPoolExecutor 創建一個線程池。executor.map 函數將 square_number 函數分布到線程中，以并行方式處理數字。通過將 max_workers 設置為 5，我們允許最多 5 個線程同時運行，這應該會顯著減少總處理時間。請隨意調整 max_workers 參數，以找到特定用例的最佳線程數。
何時使用多線程
正如你所見，多線程可以在各種場景中提供顯著的速度提升。但它并不適用于所有任務。以下是多線程特別有益的一些典型用例：
I/O 綁定任務：
- 文件 I/O：讀取和寫入文件，特別是處理大文件或多個文件時。
- 網絡 I/O：同時處理多個網絡連接，例如網絡抓取、下載文件或處理 web 服務器中的請求。
- 數據庫操作：執行 I/O 綁定的數據庫查詢，例如獲取或更新大型數據集。
并發任務：
- 實時數據處理：實時處理來自多個傳感器或流的數據，例如在 IoT 應用中。
- GUI 應用程序：通過在后臺運行耗時任務，保持用戶界面的響應性。
獨立任務的并行處理：
- 批量處理：處理大量可以并行執行的獨立任務，例如圖像處理或數據轉換任務。
- 模擬：同時運行多個模擬或蒙特卡洛實驗。
何時不使用多線程
雖然多線程可以提供顯著的速度提升，但它并不總是每個問題的最好解決方案。以下是它可能不適用的一些場景：
- CPU 綁定任務：如果任務嚴重依賴 CPU 并且不涉及太多等待(如純數學計算)，使用 multiprocessing 模塊創建單獨的進程可能更有效。
- 全局解釋器鎖 (GIL)：在 CPython 中，全局解釋器鎖可能會限制多線程在 CPU 綁定任務中的性能提升。在這種情況下，多進程或使用沒有 GIL 的實現，如 Jython 或 IronPython，可能更有效。
- 復雜的共享狀態：跨多個線程管理復雜的共享狀態可能會引入與競態條件、死鎖和線程安全性相關的挑戰和錯誤。
通過了解任務的性質和潛在瓶頸，你可以決定多線程是否是應用程序的適當解決方案。
專業提示 — 使用裝飾器
裝飾器可以用來以更優雅和可重用的方式為函數添加多線程。裝飾器是一個函數，它接受另一個函數并擴展其行為，而不需要顯式修改它。
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
# Decorator to add multithreading
def multithreaded(max_workers=5):
def decorator(func):
def wrapper(*args, **kwargs):
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_args = {executor.submit(func, arg): arg for arg in args[0]}
results = []
for future in as_completed(future_to_args):
arg = future_to_args[future]
try:
result = future.result()
except Exception as exc:
print(f'{arg} generated an exception: {exc}')
else:
results.append(result)
return results
return wrapper
return decorator
# Function to square a number
@multithreaded(max_workers=5)
def square_number(number):
time.sleep(1) # Simulate a time-consuming task
return number * number
# List of numbers to process
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Using the decorated function
start_time = time.time()
squared_numbers = square_number(numbers)
end_time = time.time()
print("Squared numbers:", squared_numbers)
print("Time taken:", end_time - start_time, "seconds")使用裝飾器處理多線程不僅簡化了代碼，還使其更可重用和更清晰。你可以輕松地將 @multithreaded 裝飾器應用于任何需要并行執行的函數，為優化你的 Python 代碼提供了一種靈活而強大的方式。
結論
多線程是優化 Python 中 for 循環的強大工具，特別是對于 I/O 綁定和并發任務。通過利用 concurrent.futures 模塊，你可以顯著減少處理時間并提高程序的效率。然而，評估你的特定用例以確定多線程是否是最佳方法至關重要，特別是當你處理 CPU 綁定任務或復雜的共享狀態時。通過仔細考慮和實施，多線程可以大大增強你的應用程序的性能。
