Python 多工與多執行緒：平行計算技術比較

– 12 分鐘閱讀

多工處理和多執行緒是提升程式效能的關鍵技術，尤其在 I/O 密集型任務中，能有效縮短程式執行時間。多工處理透過建立多個獨立程式，各自擁有獨立記憶體空間，適用於 CPU 密集型任務，能充分利用多核心處理器。多執行緒則是在單一程式內建立多個執行緒，分享記憶體空間，適用於 I/O 密集型任務，能有效減少程式建立和切換的開銷。然而，Python 的全域性直譯器鎖（GIL）限制了多執行緒在多核心處理器上的效能提升。選擇多工或多執行緒取決於任務型別和系統資源，需考量 CPU 和 I/O 密集程度，以及記憶體使用情況。

多工處理與多執行緒

在電腦科學中，多工處理（Multiprocessing）和多執行緒（Multithreading）是兩種常用的平行計算技術。這兩種技術可以提高電腦的效率和吞吐量，但它們之間存在著明顯的差異。

多工處理

多工處理是指在一個電腦系統中，同時執行多個程式（Process）的技術。每個程式都有自己的記憶體空間和系統資源，彼此之間不分享記憶體。多工處理可以透過多個CPU核心或多個電腦節點來實作。

在多工處理中，當一個程式完成任務後，可以將結果放入一個佇列（Queue）中，然後另一個程式可以從佇列中取出結果繼續處理。這樣可以實作多個程式之間的通訊和協作。

多執行緒

多執行緒是指在一個程式中，同時執行多個執行緒（Thread）的技術。執行緒是程式中的一個執行單元，分享同一塊記憶體空間和系統資源。多執行緒可以提高電腦的效率和回應速度。

在多執行緒中，當一個執行緒完成任務後，可以將結果放入一個佇列中，然後另一個執行緒可以從佇列中取出結果繼續處理。這樣可以實作多個執行緒之間的通訊和協作。

比較

多工處理和多執行緒都是平行計算技術，但它們之間存在著明顯的差異：

  • 多工處理：每個程式都有自己的記憶體空間和系統資源，彼此之間不分享記憶體。

  • 多執行緒：所有執行緒分享同一塊記憶體空間和系統資源。

  • 多工處理：需要建立多個程式，然後透過佇列或其他方式實作程式之間的通訊和協作。

  • 多執行緒：需要建立多個執行緒，然後透過佇列或其他方式實作執行緒之間的通訊和協作。

  • 多工處理：適合於大規模的計算任務，需要多個CPU核心或多個電腦節點來實作。

  • 多執行緒：適合於小規模的計算任務，需要快速回應和高效率的計算。

實踐應用

在實踐應用中，多工處理和多執行緒都被廣泛使用：

  • 多工處理：在大規模的計算任務中，例如科學計算、資料分析等。
  • 多執行緒：在小規模的計算任務中，例如網頁瀏覽、遊戲等。
程式碼範例

以下是Python中使用多工處理和多執行緒的範例：

import multiprocessing
import threading
from queue import Queue

# 多工處理
def worker(num, queue):
    result = num * 2
    queue.put(result)

if __name__ == '__main__':
    queue = multiprocessing.Manager().Queue()
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i, queue))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    results = []
    while not queue.empty():
        results.append(queue.get())
    print(results)

# 多執行緒
def worker(num, queue):
    result = num * 2
    queue.put(result)

threads = []
queue = Queue()
for i in range(5):
    t = threading.Thread(target=worker, args=(i, queue))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
results = []
while not queue.empty():
    results.append(queue.get())
print(results)

內容解密

在上述範例中，我們使用了Python的multiprocessing和threading模組來實作多工處理和多執行緒。multiprocessing模組提供了建立多個程式的功能，而threading模組提供了建立多個執行緒的功能。

我們定義了一個worker函式，該函式接收一個數字和一個佇列作為引數。該函式計算數字的兩倍，並將結果放入佇列中。

在多工處理的範例中，我們建立了5個程式，每個程式都執行worker函式。我們使用了multiprocessing.Manager().Queue()來建立一個可以被多個程式分享的佇列。

在多執行緒的範例中，我們建立了5個執行緒，每個執行緒都執行worker函式。我們使用了Queue類別來建立一個可以被多個執行緒分享的佇列。

最後，我們從佇列中取出結果，並列印預出來。

圖表翻譯

以下是上述範例的流程圖：

  flowchart TD
    A[開始] --> B[建立程式/執行緒]
    B --> C[執行worker函式]
    C --> D[計算結果]
    D --> E[放入佇列]
    E --> F[取出結果]
    F --> G[列印結果]
    G --> H[結束]

在這個流程圖中，我們可以看到多工處理和多執行緒的流程是相似的。首先，我們建立了多個程式或執行緒，每個程式或執行緒都執行worker函式。然後，我們計算結果，並將結果放入佇列中。最後，我們從佇列中取出結果，並列印預出來。

使用 Python 的 multiprocessing 模組進行平行計算

Python 的 multiprocessing 模組提供了一種方式來進行平行計算，透過建立多個程式（process）來執行任務。以下是使用 multiprocessing 模組進行平行計算的範例。

使用 Queue 進行資料傳遞

multiprocessing 模組中的 Queue 類別提供了一種方式來進行程式間的資料傳遞。以下是使用 Queue 進行資料傳遞的範例：

import multiprocessing as mp

def worker(queue, data):
    result = data * 2
    queue.put(result)

if __name__ == '__main__':
    queue = mp.Queue()
    data = [1, 2, 3, 4, 5]

    processes = []
    for d in data:
        p = mp.Process(target=worker, args=(queue, d))
        processes.append(p)
        p.start()

    results = []
    for _ in range(len(data)):
        result = queue.get()
        results.append(result)

    for p in processes:
        p.join()

    print(results)

在這個範例中，我們建立了一個 Queue 物件，並將其傳遞給每個程式。每個程式將計算結果放入 Queue 中，然後主程式可以從 Queue 中取出結果。

使用 Pipe 進行資料傳遞

multiprocessing 模組中的 Pipe 類別提供了一種方式來進行程式間的資料傳遞。以下是使用 Pipe 進行資料傳遞的範例：

import multiprocessing as mp

def worker(conn, data):
    result = data * 2
    conn.send(result)
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = mp.Pipe()
    data = [1, 2, 3, 4, 5]

    processes = []
    for d in data:
        p = mp.Process(target=worker, args=(child_conn, d))
        processes.append(p)
        p.start()

    results = []
    for _ in range(len(data)):
        result = parent_conn.recv()
        results.append(result)

    for p in processes:
        p.join()

    print(results)

在這個範例中，我們建立了一個 Pipe 物件，並將其傳遞給每個程式。每個程式將計算結果傳遞給主程式，然後主程式可以接收結果。

使用 Pool 進行平行計算

multiprocessing 模組中的 Pool 類別提供了一種方式來進行平行計算。以下是使用 Pool 進行平行計算的範例：

import multiprocessing as mp

def worker(data):
    result = data * 2
    return result

if __name__ == '__main__':
    data = [1, 2, 3, 4, 5]

    with mp.Pool() as pool:
        results = pool.map(worker, data)

    print(results)

在這個範例中，我們建立了一個 Pool 物件，並將其用於進行平行計算。每個程式將計算結果傳回給主程式，然後主程式可以接收結果。

多程式平行處理溫度轉換

在進行多程式平行處理時，需要注意管道（Pipe）通訊的使用，以避免不同程式之間的資料衝突。以下是一個使用 Python 的 multiprocessing 模組進行溫度轉換的範例：

原始碼分析

原始碼中定義了一個函式 to_celsius，它接收一個子程式管道 (child_pipe) 作為引數，從父程式接收一個華氏溫度，將其轉換為攝氏溫度，然後將結果發送回父程式。

def to_celsius(child_pipe: mp.Pipe):
    f = child_pipe.recv()  # 接收華氏溫度
    c = (f - 32) * (5/9)  # 轉換為攝氏溫度
    child_pipe.send(c)  # 傳送結果回父程式

在主程式中，建立了一個多程式池 (Pool)，並建立了多個子程式來執行溫度轉換任務。

if __name__ == '__main__':
    mp.set_start_method('spawn')  # 設定多程式啟動方法
    pool_manager = mp.Manager()  # 建立多程式管理器
    
    with mp.Pool(2) as pool:  # 建立一個具有 2 個子程式的池
        parent_pipe, child_pipe = mp.Pipe()  # 建立一個管道
        
        results = []  # 用於儲存結果
        
        for input_temp in range(110, 150, 10):  # 迭代輸入溫度
            parent_pipe.send(input_temp)  # 將輸入溫度傳送給子程式
            results.append(pool.apply_async(to_celsius, args=(child_pipe,)))  # 啟動子程式
            
            print("Got {0:}".format(parent_pipe.recv()))  # 接收子程式傳送的結果
        
        parent_pipe.close()  # 關閉父程式管道
        child_pipe.close()  # 關閉子程式管道

潛在問題

在這段原始碼中，存在一個潛在問題：當多個子程式嘗試同時從父程式管道中讀取資料時，可能會導致資料衝突或損壞。這是因為多個子程式分享同一個父程式管道，如果不進行適當的同步，可能會導致不可預測的行為。

解決方案

為瞭解決這個問題，可以使用 multiprocessing 模組提供的同步工具，例如 Lock 或 Queue，來確保子程式之間的資料存取是安全的。另外，也可以考慮使用其他的平行處理方法，例如使用 concurrent.futures 模組，來避免管道通訊的複雜性。

圖表翻譯
  flowchart TD
    A[主程式] -->|建立管道|> B[父程式管道]
    A -->|建立子程式|> C[子程式 1]
    A -->|建立子程式|> D[子程式 2]
    C -->|接收資料|> B
    D -->|接收資料|> B
    B -->|傳送結果|> C
    B -->|傳送結果|> D

內容解密

在這個範例中，我們使用 multiprocessing 模組來進行多程式平行處理。主程式建立了一個管道和多個子程式，然後將輸入溫度傳送給子程式。子程式接收輸入溫度，進行溫度轉換，然後將結果發送回父程式。父程式接收結果，並列印預出來。

然而，在這個過程中，存在一個潛在問題：當多個子程式嘗試同時從父程式管道中讀取資料時，可能會導致資料衝突或損壞。為瞭解決這個問題，可以使用同步工具或其他平行處理方法來確保資料交換的安全性。

多程式間的溝通與同步

在多程式環境中，程式間的溝通和同步是非常重要的。Python 的 multiprocessing 模組提供了管道（Pipe）和鎖定（Lock）等機制來實作程式間的溝通和同步。

管道（Pipe）

管道是一種半雙工的溝通方式，允許兩個程式之間進行資料交換。管道可以用於父子程式之間的溝通，也可以用於兄弟程式之間的溝通。

import multiprocessing as mp

def child_process(pipe):
    # 從管道接收資料
    data = pipe.recv()
    print(f"Child process received: {data}")

    # 將資料發送回父程式
    pipe.send("Hello from child process!")

if __name__ == "__main__":
    # 建立管道
    parent_conn, child_conn = mp.Pipe()

    # 建立子程式
    p = mp.Process(target=child_process, args=(child_conn,))

    # 啟動子程式
    p.start()

    # 將資料傳送給子程式
    parent_conn.send("Hello from parent process!")

    # 從管道接收資料
    data = parent_conn.recv()
    print(f"Parent process received: {data}")

    # 等待子程式結束
    p.join()

鎖定（Lock）

鎖定是一種同步機制，允許只有一個程式可以存取分享資源。鎖定可以用於防止多個程式同時存取分享資源，從而避免資料損壞或其他同步問題。

import multiprocessing as mp

def worker(lock, num):
    with lock:
        print(f"Worker {num} is working...")
        # 進行一些工作...
        print(f"Worker {num} finished.")

if __name__ == "__main__":
    # 建立鎖定
    lock = mp.Lock()

    # 建立多個工作者程式
    workers = []
    for i in range(5):
        p = mp.Process(target=worker, args=(lock, i))
        workers.append(p)
        p.start()

    # 等待所有工作者程式結束
    for p in workers:
        p.join()

Semaphore

Semaphore是一種同步機制，允許控制多個程式存取分享資源的數量。Semaphore可以用於防止太多程式同時存取分享資源，從而避免資料損壞或其他同步問題。

import multiprocessing as mp
import time

def worker(sem, num):
    sem.acquire()
    try:
        print(f"Worker {num} is working...")
        time.sleep(2)
        print(f"Worker {num} finished.")
    finally:
        sem.release()

if __name__ == "__main__":
    # 建立Semaphore
    sem = mp.Semaphore(3)

    # 建立多個工作者程式
    workers = []
    for i in range(10):
        p = mp.Process(target=worker, args=(sem, i))
        workers.append(p)
        p.start()

    # 等待所有工作者程式結束
    for p in workers:
        p.join()

範例：溫度轉換

以下範例示範如何使用管道和鎖定來實作父子程式之間的溝通和同步：

import multiprocessing as mp

def to_celsius(child_pipe, child_lock):
    child_lock.acquire(blocking=False)
    try:
        f = child_pipe.recv()
        c = (f - 32) * (5/9)
        print(f"{f} Fahrenheit is {c} Celsius")
    finally:
        child_lock.release()

if __name__ == "__main__":
    # 建立管道
    parent_conn, child_conn = mp.Pipe()

    # 建立鎖定
    child_lock = mp.Lock()

    # 建立子程式
    p = mp.Process(target=to_celsius, args=(child_conn, child_lock))

    # 啟動子程式
    p.start()

    # 將資料傳送給子程式
    parent_conn.send(100)

    # 等待子程式結束
    p.join()

這個範例示範如何使用管道和鎖定來實作父子程式之間的溝通和同步，從而避免資料損壞或其他同步問題。

多程式計算與同步鎖

在進行多程式計算時，為了確保資料的一致性和避免競爭條件，同步鎖（Lock）是一種非常重要的機制。以下將展示如何使用Python的multiprocessing模組來建立多程式環境，並利用同步鎖來保護分享資源。

多程式環境建立

首先，我們需要建立一個多程式環境。Python的multiprocessing模組提供了Pool類別來簡化這個過程。下面的程式碼展示瞭如何建立一個包含2個程式的池（Pool）。

import multiprocessing as mp

if __name__ == '__main__':
    # 設定程式啟動方法為'spawn'
    mp.set_start_method('spawn')
    
    # 建立一個程式池管理器
    pool_manager = mp.Manager()
    
    # 建立一個包含2個程式的池
    with mp.Pool(2) as pool:
        # 進行後續的多程式計算
        pass

程式間通訊與同步鎖

在多程式計算中，程式間通訊是一個重要的議題。Python的multiprocessing模組提供了管道（Pipe）來實作程式間通訊。另外，為了保護分享資源，同步鎖（Lock）是必不可少的。

# 建立一個管道用於程式間通訊
parent_pipe, child_pipe = mp.Pipe()
    
# 建立一個同步鎖
child_lock = pool_manager.Lock()

多程式計算與同步鎖應用

現在，我們可以進行多程式計算，並利用同步鎖來保護分享資源。假設我們有一個函式to_celsius，它需要接受一個值並進行計算。

def to_celsius(value):
    # 進行計算
    result = (value - 32) * 5.0/9.0
    return result

# 多程式計算
results = []
for i in range(110, 150, 10):
    # 將資料傳送給子程式
    parent_pipe.send(i)
    
    # 利用apply_async進行非同步計算
    results.append(pool.apply_async(to_celsius, args=(child_pipe, child_lock)))

保護分享資源

在上述程式碼中，我們使用同步鎖來保護分享資源。在子程式中，當接收到資料後，應該立即對分享資源進行鎖定，以避免其他程式幹擾。

# 子程式中
child_lock.acquire(blocking=False)
try:
    # 處理分享資源
    child_pipe.send(c)
finally:
    child_lock.release()

最終結果處理

最後，需要等待所有非同步計算任務完成，並取出結果。

# 等待所有任務完成
for result in results:
    print(result.get())

多工處理與多執行緒的比較

在進行網路掃描時，需要考慮到效率和速度。為了達到這個目標，常常會使用多工處理（Multiprocessing）或多執行緒（Multithreading）的技術。下面是一個簡單的範例，展示瞭如何使用這兩種技術來實作一個TCP埠掃描器。

多工處理的實作

首先，我們來看一下使用多工處理的實作。這個範例使用了Python的multiprocessing模組來建立多個程式，每個程式負責掃描一段埠範圍。

import multiprocessing
import socket
import time

def check_port(host, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(1.0)
    try:
        sock.connect((host, port))
        print(f"Port {port} is open")
    except socket.error:
        pass
    finally:
        sock.close()

def scan_ports(host, start_port, end_port):
    processes = []
    for port in range(start_port, end_port + 1):
        p = multiprocessing.Process(target=check_port, args=(host, port))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

if __name__ == "__main__":
    host = "example.com"
    start_port = 80
    end_port = 100
    start_time = time.time()
    scan_ports(host, start_port, end_port)
    print(f"Scan completed in {time.time() - start_time} seconds")

多執行緒的實作

接下來，我們來看一下使用多執行緒的實作。這個範例使用了Python的threading模組來建立多個執行緒，每個執行緒負責掃描一段埠範圍。

import threading
import socket
import time

def check_port(host, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(1.0)
    try:
        sock.connect((host, port))
        print(f"Port {port} is open")
    except socket.error:
        pass
    finally:
        sock.close()

def scan_ports(host, start_port, end_port):
    threads = []
    for port in range(start_port, end_port + 1):
        t = threading.Thread(target=check_port, args=(host, port))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

if __name__ == "__main__":
    host = "example.com"
    start_port = 80
    end_port = 100
    start_time = time.time()
    scan_ports(host, start_port, end_port)
    print(f"Scan completed in {time.time() - start_time} seconds")

比較

兩個範例都能夠完成TCP埠掃描的任務，但是它們之間有一些差異。

  • 多工處理的實作使用了多個程式，每個程式負責掃描一段埠範圍。這個方法可以充分利用多核CPU的資源，提高掃描速度。
  • 多執行緒的實作使用了多個執行緒，每個執行緒負責掃描一段埠範圍。這個方法可以減少程式建立和切換的開銷，但是由於Python的全域性直譯器鎖（GIL），多執行緒可能不能夠充分利用多核CPU的資源。

在實際應用中，需要根據具體的情況選擇合適的方法。如果需要充分利用多核CPU的資源，多工處理可能是一個更好的選擇。如果需要減少程式建立和切換的開銷，多執行緒可能是一個更好的選擇。

從效能最佳化視角來看，多工處理（Multiprocessing）和多執行緒（Multithreading）是提升程式效能的利器，但並非所有情況都適用。本文深入比較了這兩種技術的差異、應用場景以及潛在的效能陷阱。多工處理透過利用多個CPU核心實作真正的平行，適用於CPU密集型任務，例如大規模資料處理和科學計算。然而，程式間的通訊成本較高，需要謹慎設計資料交換策略。多執行緒則更輕量，適用於I/O密集型任務，例如網路操作和檔案讀寫。但受限於全域性直譯器鎖（GIL），Python的多執行緒在CPU密集型任務中效能提升有限。此外，分享記憶體也增加了同步和鎖定的複雜度，需注意避免競爭條件和死結。對於Python開發者，concurrent.futures 模組提供更簡潔易用的介面，能有效管理多程式和多執行緒，值得深入研究。玄貓認為，選擇合適的平行處理技術需根據任務型別、資源限制和開發成本綜合考量，才能最大化效能提升。未來，隨著Python生態的發展，預計會有更多高階工具和技術出現，進一步簡化平行計算的開發流程，並提升效能。

玄貓

玄貓

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