Python 多執行緒加速 CPU 密集型任務

– 8 分鐘閱讀

Python 的 GIL 限制了多執行緒在 CPU 密集型任務中的真正平行化，但透過 C 擴充套件以及 I/O 操作的特性，仍然可以提升效能。本文以 hashlib 和 NumPy 等實際案例，示範如何利用多執行緒加速資料處理。此外，結合 asyncio 與 ThreadPoolExecutor，更能有效管理非同步任務，提升程式碼執行效率。文章也提供流程圖和序列圖，解析程式架構，並探討如何在 Windows 環境下處理標準輸入，以及進階應用、安全考量和最佳實務。

掌握 Python 多執行緒加速 CPU 密集任務

在 Python 的世界裡，提升 CPU 密集型任務的效率一直是個熱門話題。多執行緒程式設計常常被視為一把利器，然而，全域性直譯器鎖（GIL）的存在又讓它蒙上了一層神秘的面紗。本文將揭開這層面紗，探討 Python 多執行緒的實用技巧，並結合 asyncio 和 ThreadPoolExecutor，展現如何最大程度地發揮多執行緒的優勢。我們將以 hashlib 和 NumPy 為例，剖析如何在實際場景中進行效能最佳化。

Python 多執行緒與 GIL 的博弈

Python 的 GIL 機制，就像一位嚴格的交通指揮員，同一時間只允許一個執行緒持有直譯器的控制權。這使得真正的平行執行在多執行緒場景下難以實作。然而，對於 I/O 密集型任務，多執行緒仍然能夠提升效率，因為執行緒在等待 I/O 操作完成時會釋放 GIL，允許其他執行緒執行。

而對於 CPU 密集型任務，情況就略顯複雜。如果任務涉及到 C 擴充套件，例如 hashlib 和 NumPy 的部分操作，它們在執行過程中可能會釋放 GIL，從而允許多個執行緒平行執行，提升效能。

hashlib 多執行緒加密：守護資料安全

資料安全是現代應用的根本，hashlib 函式庫提供了豐富的加密演算法，例如 SHA512 和 scrypt。當處理大量資料時，多執行緒可以顯著提升加密效率。

以下程式碼展示瞭如何使用多執行緒加速 SHA512 雜湊：

import hashlib
import os
from concurrent.futures import ThreadPoolExecutor
import time

def hash_data(data: bytes) -> str:
    """使用 SHA512 演算法雜湊資料"""
    return hashlib.sha512(data).hexdigest()

def multithreaded_hashing(data_chunks: list) -> list:
    """多執行緒雜湊資料塊"""
    with ThreadPoolExecutor() as executor:
        results = list(executor.map(hash_data, data_chunks))
    return results

# 模擬 20MB 資料
data = os.urandom(20 * 1024 * 1024)
chunk_size = 1024 * 1024
data_chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]

start_time = time.time()
hashed_chunks = multithreaded_hashing(data_chunks)
end_time = time.time()

print(f"多執行緒雜湊耗時：{end_time - start_time:.2f} 秒")

# 單執行緒雜湊對比
start_time = time.time()
for chunk in data_chunks:
    hash_data(chunk)
end_time = time.time()
print(f"單執行緒雜湊耗時：{end_time - start_time:.2f} 秒")

內容解密：

這段程式碼將 20MB 資料分割成多個塊，利用 ThreadPoolExecutor 建立執行緒池，平行計算每個資料塊的 SHA512 雜湊值。透過與單執行緒版本對比，可以明顯看出多執行緒帶來的效能提升。

asyncio 與 ThreadPoolExecutor 的強強聯手

asyncio 是 Python 的非同步程式設計框架，而 ThreadPoolExecutor 則提供了執行緒池管理功能。將兩者結合，可以更好地處理 CPU 密集型任務。

讓我們以 scrypt 演算法為例，展示如何結合 asyncio 和 ThreadPoolExecutor 提升密碼雜湊效率：

import asyncio
import hashlib
import os
from concurrent.futures import ThreadPoolExecutor
import random
import string

def generate_password(length: int = 10) -> bytes:
    """生成隨機密碼"""
    characters = string.ascii_letters + string.digits
    return ''.join(random.choice(characters) for _ in range(length)).encode()

def hash_password(password: bytes) -> bytes:
    """使用 scrypt 演算法雜湊密碼"""
    salt = os.urandom(16)
    return hashlib.scrypt(password, salt=salt, n=16384, r=8, p=1)

async def main():
    passwords = [generate_password() for _ in range(100)]

with ThreadPoolExecutor() as executor:
    loop = asyncio.get_running_loop()
    tasks = [loop.run_in_executor(executor, hash_password, password) for password in passwords]
    results = await asyncio.gather(*tasks)

# 處理結果
    print(f"已雜湊 {len(results)} 個密碼")

asyncio.run(main())

內容解密：

這段程式碼首先生成 100 個隨機密碼，然後利用 ThreadPoolExecutor 建立執行緒池，透過 loop.run_in_executor 將密碼雜湊任務提交到執行緒池中執行。asyncio.gather 協程用於等待所有任務完成，並收集結果。

NumPy 與多執行緒的微妙關係

NumPy 在進行某些操作時會釋放 GIL，因此多執行緒可以提升 NumPy 程式碼的效能。

以下程式碼展示瞭如何使用多執行緒計算大型矩陣每行的平均值：

import asyncio
import numpy as np
from concurrent.futures import ThreadPoolExecutor
import time

def calculate_row_mean(matrix: np.ndarray, row_index: int) -> float:
    """計算矩陣指定行的平均值"""
    return np.mean(matrix[row_index])

async def main():
    matrix = np.random.rand(1000, 10000)  # 建立一個大型矩陣

with ThreadPoolExecutor() as executor:
    loop = asyncio.get_running_loop()
    tasks = [loop.run_in_executor(executor, calculate_row_mean, matrix, i) for i in range(matrix.shape[0])]
    results = await asyncio.gather(*tasks)

# 處理結果
    print(f"已計算 {len(results)} 行的平均值")

asyncio.run(main())

內容解密：

這段程式碼建立一個 1000x10000 的隨機矩陣，利用 ThreadPoolExecutor 平行計算每行的平均值。由於 NumPy 的 np.mean 操作會釋放 GIL，多執行緒可以有效提升計算效率。

效能最佳化之道：多管齊下

在 Python 中，提升 CPU 密集型任務的效率需要多管齊下。除了多執行緒，還可以考慮使用多行程、Cython 或其他最佳化技巧。選擇合適的方案需要根據具體場景進行評估和測試。

  graph LR
    C[C]
    A[資料分塊] --> B(多執行緒處理)
    B --> C{結果合併}

圖表翻譯：

這張流程圖展示了多執行緒資料處理的基本流程，包括資料分塊、多執行緒處理和結果合併三個步驟。

  sequenceDiagram
    participant Main
    participant ThreadPool
    Main->>ThreadPool: 提交任務
    activate ThreadPool
    ThreadPool-->>Main: 傳回結果
    deactivate ThreadPool

圖表翻譯：

這張序列圖展示了主執行緒與執行緒池之間的互動過程，主執行緒提交任務給執行緒池，執行緒池處理完成後傳回結果。

進階功能開發

當你掌握了基本的多執行緒技術後，可以進一步探索更複雜的應用案例，例如：

  • 非同步 I/O 操作：運用 asyncio 的非同步 I/O 功能，最佳化網路請求、檔案操作等。
  • 平行計算：使用 multiprocessing 模組，結合多執行緒和多程式，實作更高階的平行計算。
  • 最佳化資料結構：根據具體任務需求，選擇合適的資料結構，如 NumPy 陣列或 Dask 進行高效的資料處理。

例項應用：高效資料分析

假設我們需要對大量金融交易資料進行快速的統計分析。我們可以使用多執行緒和 asyncio 來最佳化以下流程：

  1. 資料讀取：運用多執行緒平行從多個檔案中讀取交易資料。
  2. 資料處理：利用 asyncio 非同步資料處理，進行資料清洗、過濾和轉換。
  3. 統計分析：使用 NumPy 和多執行緒計算資料統計指標，如均值、中位數等。
  4. 結果輸出：將分析結果以圖表形式呈現，提供快速的視覺化反饋。

安全考量與最佳實踐

在使用多執行緒和 asyncio 時，也需考慮以下安全事項：

  • 並發控制：避免多執行緒競爭，使用鎖機制（如 asyncio.Lock）來保護分享資源。
  • 錯誤處理：精心設計錯誤處理機制，確保在多執行緒環境下也能有效地捕捉和處理異常。
  • 資源管理：合理管理執行緒資源，避免資源耗盡或記憶體洩漏。

Python asyncio 中處理非同步輸入的技巧

在 Python 的 asyncio 函式庫中，處理非同步輸入是構建高效命令列應用程式的關鍵。本文將深入解析如何利用 asyncio.StreamReader 和 asyncio.StreamReaderProtocol 建立非同步的標準輸入讀取器，並結合 asyncio.create_task() 來處理耗時的任務，實作流暢的命令列互動。

基礎架構與原理

核心問題是如何在 asyncio 環境中，同時處理來自標準輸入的資料和後面的延遲任務，而不會阻塞主迴圈。

  • create_stdin_reader() 協程:

    • 使用 asyncio.StreamReader 和 asyncio.StreamReaderProtocol 建立一個非同步的標準輸入讀取器。
    • 此讀取器可以 asynchronously 接收來自標準輸入的資料。

  • delay() 協程:

    • 模擬了需要耗時的任務，例如檔案操作、網路請求等。
    • 在任務開始和結束時印出訊息，提供診斷資訊。

  • main() 協程:

    • 使用 create_stdin_reader() 建立的讀取器，持續讀取使用者輸入。
    • 將讀取的每一行輸入轉換為整數。
    • 使用 asyncio.create_task() 將 delay() 任務放入事件迴圈中併發執行，而不會阻塞主迴圈。
    • 加入錯誤處理機制，避免非整數輸入造成程式當機。

圖表分析

以下是流程圖和序列圖的解析：

流程圖:

這張圖清晰地展示了程式的主要執行流程：

  • 讀取輸入: 主迴圈持續從標準輸入讀取一行輸入。
  • 驗證輸入: 將輸入轉換為整數，並檢查是否有效。
  • 建立延遲任務: 如果輸入有效，建立 delay() 協程任務。
  • 事件迴圈: delay() 任務被放入事件迴圈中，等待執行。

序列圖:

這張圖更詳細地呈現了主迴圈、讀取器和延遲任務之間的互動順序：

  • 主迴圈 -> 讀取器: 主迴圈從讀取器接收輸入值。
  • 讀取器 -> 主迴圈: 讀取器將輸入值傳回主迴圈進行驗證。
  • 主迴圈 -> 延遲任務: 驗證透過後，主迴圈建立 delay() 任務。
  • 延遲任務 -> 事件迴圈: delay() 任務被新增到事件迴圈中。
  • 事件迴圈 -> 延遲任務: 事件迴圈呼叫 delay() 任務，執行其任務。
  • 延遲任務 -> 事件迴圈: 任務完成後， delay() 任務傳回事件迴圈。

環境設定與準備

需要注意的是，在 Windows 系統上，connect_read_pipe() 與 sys.stdin 的搭配存在相容性問題。

  • 解決方案:
    • 使用獨立的執行緒，在後臺使用 sys.stdin.readline() 來讀取輸入。
    • 採用其他跨平臺的解決方案，例如 asyncio.open_connection() 來建立 Socket 連線，並從連線中讀取資料。

另外，建議將終端設定為 raw 模式，以避免輸入回顯和特殊鍵的幹擾，這會提升使用者經驗。

進階應用

除了基本的輸入處理，你還可以新增更多功能，例如：

  • 日誌記錄: 將輸入和任務執行結果記錄到檔案或日誌。
  • 錯誤回報: 提供更詳細的錯誤資訊，並通知使用者。
  • 進度條: 在延遲任務執行期間，顯示進度條，讓使用者清楚瞭解執行狀態。

安全與最佳實踐

  • 錯誤處理: 徹底處理錯誤，以優雅地處理意外輸入或錯誤。

  • 資源管理: 有效管理資源，例如檔案描述符、連線等，防止資源耗盡。

  • 效能監控: 監視程式的效能，避免因過度處理而導致效能瓶頸。

縱觀 Python 多執行緒程式設計的技術脈絡，本文深入探討瞭如何在 CPU 密集型任務中有效利用多執行緒提升效能。從 GIL 的限制到 hashlib、asyncio、ThreadPoolExecutor 以及 NumPy 的整合運用，我們揭示了多執行緒程式設計的實用技巧和最佳實踐。尤其是在結合 asyncio 與 ThreadPoolExecutor 處理 scrypt 加密和 NumPy 矩陣計算的案例中，更展現了其在實際應用中的優勢。然而，Python 多執行緒並非效能提升的萬靈丹，需根據任務特性謹慎評估，並結合多行程、Cython 等其他最佳化策略，才能最大化程式效能。未來，隨著 Python 生態的持續發展，預期會有更多針對多執行緒程式設計的最佳化方案出現，進一步提升 Python 在 CPU 密集型任務中的表現。同時，針對 asyncio 非同步輸入處理的深入解析，也為構建高互動性命令列應用程式提供了實用，建議開發者關注並探索更進階的應用場景。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。