使用 Python 的多執行緒指南｜從初學者到實戰全面掌握

By 佐川 直弘 | Naohiro Sagawa (運営者)

※記事内に広告を含みます。Amazonアソシエイトとして適格販売により収入を得ています。

1. 前言

Python 是一種因其簡潔易懂的語法與豐富的函式庫而受到初學者到進階使用者廣泛使用的程式語言。其中，多執行緒技術在特定情境下能大幅提升處理效率，是一項非常重要的技巧。

為什麼要在 Python 中使用多執行緒

隨著電腦效能的提升，程式所需處理的資料量與處理速度的需求也越來越高。以下幾種情境特別適合活用多執行緒：

• 大量資料處理：當需要從資料庫讀取資料或處理大量檔案時，可透過平行化來縮短處理時間。
• I/O 操作效率提升：對於頻繁進行檔案讀寫或網路通訊的程式，可以減少等待時間，提高整體效率。
• 需要即時性的應用：在開發遊戲或使用者介面等情況中，通常需要同時處理多個任務，因此多執行緒是不可或缺的。

多執行緒的優點與挑戰

優點

1. 加快處理速度：多個執行緒可以同時執行，讓任務分工更有效率。
2. 有效利用資源：即使某些執行緒在等待中，其他執行緒仍能使用 CPU 資源持續運作。

挑戰

1. 全域直譯器鎖（GIL）限制：由於 GIL 的存在，在某些情況下 Python 的多執行緒效能會受到限制。
2. 除錯困難：執行緒之間可能發生資源競爭或死結等問題，導致除錯時間增加。

本文目的

本篇文章將深入介紹在 Python 中實作多執行緒的基本概念與實際方法，並透過範例說明常見應用與注意事項。內容設計上特別考量初學者與中階開發者的需求，讓讀者能逐步理解與應用於實務開發中，歡迎閱讀至最後！

2. 多執行緒與多進程的比較

在程式設計中，多執行緒與多進程都是實現平行處理的關鍵技術，但它們各有特性與適用場景。本節將深入比較兩者的差異，並說明如何在 Python 中正確使用。

執行緒與進程的基本差異

什麼是執行緒

執行緒是在單一進程中執行的平行處理單位，會共享相同的記憶體空間，因此資料傳遞效率高。

• 特點：
• 共享記憶體空間
• 啟動快速、執行輕量
• 容易進行資料共享

什麼是進程

進程是具有獨立記憶體空間的執行單位。由於每個進程擁有自己的資源，因此彼此影響較小。

• 特點：
• 擁有獨立記憶體空間
• 啟動較慢、較為耗資源
• 資料共享需要額外機制

Python 中的 GIL（全域直譯器鎖）影響

Python 中存在名為 GIL（Global Interpreter Lock）的限制機制，這會讓程式在任一時刻只能由一個執行緒執行 Python bytecode。這使得即使使用多執行緒，也可能無法充分發揮多核心 CPU 的效能。

• 容易受 GIL 影響的情況：
• 需要大量 CPU 的運算處理（如數值計算、影像處理）
• 不太受 GIL 影響的情況：
• 以 I/O 操作為主的任務（如網路通訊、檔案處理）

多執行緒與多進程的選用方式

選擇多執行緒的情況

• 適用場景：
• 需要進行大量 I/O 操作的程式
• 需平行執行多個輕量級任務時
• 範例： 網頁爬蟲、同時下載多個檔案

選擇多進程的情況

• 適用場景：
• 需要大量 CPU 運算的任務
• 想要避開 GIL 限制的情況
• 範例： 機器學習模型訓練、影像處理

Python 中簡單的比較範例

以下是使用 Python 的 threading 與 multiprocessing 模組，實現簡單平行處理的範例程式碼。

3. 執行緒與進程的基本概念

為了正確理解與應用多執行緒與多進程，了解它們各自的基本機制與特性是非常重要的。本節將說明執行緒與進程的運作方式，以及在什麼情況下適合使用它們。

執行緒的基本概念

執行緒的角色

執行緒是指在同一個進程內部獨立執行的處理流程。由於多個執行緒會共享同一記憶體空間，資料的共享與傳遞變得更加順暢。

• 特點：
• 在單一進程中運作的輕量單位。
• 共享記憶體空間，資料交換速度快。
• 需要處理執行緒間的同步與競爭問題。

執行緒的優點與挑戰

• 優點：
• 記憶體使用效率高。
• 啟動與切換速度快。
• 挑戰：
• 資料共享容易產生競爭或死結風險。
• 在 Python 中會受到 GIL 限制，不適合進行高 CPU 負載的處理。

進程的基本概念

進程的角色

進程是由作業系統所分配的獨立執行環境。每個進程都有自己獨立的記憶體空間，因此彼此之間不會互相干擾。

• 特點：
• 使用完全獨立的記憶體空間。
• 具有較高的安全性與穩定性。
• 若需要與其他進程通信（IPC），實作會較為複雜。

進程的優點與挑戰

• 優點：
• 不受 GIL 影響，適合進行高 CPU 負載的處理。
• 每個進程獨立運作，穩定性更高。
• 挑戰：
• 啟動與切換成本高。
• 記憶體使用量較大。

執行緒與進程的運作比較

全域直譯器鎖（GIL）的機制

在 Python 中，GIL 控制著執行緒的執行行為。GIL 的作用是保證在任一時間點，只有一個執行緒可以執行 Python 的位元碼。雖然這能防止資料競爭，確保執行緒安全，但也會在使用多核心 CPU 時產生效能瓶頸。

• GIL 的優點：
• 防止資料競爭，確保執行緒安全。
• GIL 的缺點：
• 在高 CPU 負載的任務中，多執行緒的效能會受到限制。

執行緒與進程的選擇準則

在 Python 中進行平行處理時，可以根據以下條件來選擇使用執行緒還是進程：

• 適合使用執行緒的情況：
• 大多數處理為 I/O 等待（例如：網路通訊）。
• 希望降低記憶體使用量。
• 適合使用進程的情況：
• 需要大量 CPU 的運算（例如：數值計算）。
• 希望充分利用多核心 CPU。

4. 在 Python 中實作多執行緒

在 Python 中實作多執行緒時，會使用標準函式庫中的 threading 模組。本節將透過具體程式碼範例，說明從基本的執行緒建立到進階的控制方式。

threading 模組的基本用法

建立與執行執行緒

使用 threading 模組時，可透過 Thread 類別來建立並啟動執行緒。以下是基本範例：

import threading
import time

def print_message(message):
    print(f"開始: {message}")
    time.sleep(2)
    print(f"結束: {message}")

## 建立執行緒
thread1 = threading.Thread(target=print_message, args=("執行緒1",))
thread2 = threading.Thread(target=print_message, args=("執行緒2",))

## 啟動執行緒
thread1.start()
thread2.start()

## 等待執行緒結束
thread1.join()
thread2.join()

print("所有執行緒已完成")

執行結果說明

在這段程式碼中，兩個執行緒會同時啟動並各自執行。透過 join() 方法可以讓主執行緒等待所有子執行緒完成後再繼續執行。

使用類別實作執行緒

你也可以繼承 Thread 類別來定義更複雜的執行緒行為。

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(f"{self.name} 開始")
        time.sleep(2)
        print(f"{self.name} 結束")

## 建立執行緒
thread1 = MyThread("執行緒1")
thread2 = MyThread("執行緒2")

## 啟動執行緒
thread1.start()
thread2.start()

## 等待執行緒結束
thread1.join()
thread2.join()

print("所有執行緒已完成")

執行結果說明

在 run() 方法中定義具體的處理內容，並透過 start() 啟動執行緒。這種方式特別適合需要封裝邏輯或重複使用的情境。

執行緒之間的同步與鎖定

當多個執行緒同時操作共享資料時，可能會發生資料競爭或不一致的問題。為避免此情況，可使用 Lock 物件來實現執行緒同步。

使用鎖定的範例

import threading

lock = threading.Lock()
shared_resource = 0

def increment():
    global shared_resource
    with lock:  ## 取得鎖定
        local_copy = shared_resource
        local_copy += 1
        shared_resource = local_copy

threads = []
for i in range(5):
    thread = threading.Thread(target=increment)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"共享資源的最終值: {shared_resource}")

執行結果說明

透過 with lock 的語法可以安全地取得與釋放鎖定。在這個範例中，鎖定機制確保一次只有一個執行緒可以修改共享變數。

執行緒的超時與背景執行（Daemon）

執行緒超時

可以為 join() 方法設定超時時間，讓主執行緒只等待指定時間。

thread.join(timeout=5)

背景執行緒（Daemon Thread）

背景執行緒會在主執行緒結束時自動終止。可透過設定 daemon 屬性為 True 來建立背景執行緒。

thread = threading.Thread(target=print_message)
thread.daemon = True
thread.start()

實務中的多執行緒應用範例

以下是一個將多個檔案同時下載的平行處理範例：

import threading
import time

def download_file(file_name):
    print(f"{file_name} 開始下載")
    time.sleep(2)  ## 模擬下載時間
    print(f"{file_name} 下載完成")

files = ["file1", "file2", "file3"]

threads = []
for file in files:
    thread = threading.Thread(target=download_file, args=(file,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有檔案下載完成")

結論

本節介紹了如何在 Python 中從基本到進階地實作多執行緒功能，並結合實務應用案例。接下來的章節將更深入探討多執行緒的具體應用範例。

5. 多執行緒的實務應用範例

在 Python 中，多執行緒特別適用於 I/O 等待時間長的處理情境。本節將介紹幾個具體的應用範例，幫助你理解如何在實際專案中有效運用多執行緒。

1. 提升 Web 爬蟲的效率

從網站收集資料時，若能同時對多個 URL 發送請求，可大幅縮短整體處理時間。

範例程式碼

以下是使用 requests 函式庫與 threading 模組來進行 Web 爬蟲的範例：

import threading
import requests
import time

urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3"
]

def fetch_url(url):
    print(f"開始抓取: {url}")
    response = requests.get(url)
    print(f"完成抓取: {url}，狀態碼 {response.status_code}")

threads = []
start_time = time.time()

for url in urls:
    thread = threading.Thread(target=fetch_url, args=(url,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

end_time = time.time()
print(f"總處理時間: {end_time - start_time:.2f} 秒")

執行結果說明

透過多執行緒，同時對多個 URL 進行請求，能顯著縮短等待時間。不過，若請求數過多，需注意伺服器負載與使用規範。

2. 同時下載多個檔案

從網路下載多個檔案時，利用多執行緒可加速整體下載流程。

範例程式碼

import threading
import time

def download_file(file_name):
    print(f"{file_name} 開始下載")
    time.sleep(2)  ## 模擬下載
    print(f"{file_name} 下載完成")

files = ["file1.zip", "file2.zip", "file3.zip"]

threads = []
for file in files:
    thread = threading.Thread(target=download_file, args=(file,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有檔案下載完成")

執行結果說明

此範例中，透過多執行緒同時進行多個檔案下載，可縮短等待時間。實際應用中可搭配 urllib 或 requests 來實作真實下載。

3. 平行執行資料庫查詢

若需從資料庫查詢大量資料，透過多執行緒平行查詢可加速整體處理效率。

範例程式碼

import threading
import time

def query_database(query):
    print(f"執行查詢: {query}")
    time.sleep(2)  ## 模擬查詢
    print(f"查詢完成: {query}")

queries = ["SELECT * FROM users", "SELECT * FROM orders", "SELECT * FROM products"]

threads = []
for query in queries:
    thread = threading.Thread(target=query_database, args=(query,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有查詢已完成")

執行結果說明

這個範例展示了如何同時執行多個查詢指令，以減少等待時間。實際應用中會使用如 sqlite3 或 psycopg2 等資料庫模組來連接資料庫。

4. 平行處理影片畫面

針對影片逐格處理的任務，也能透過多執行緒來加速。

範例程式碼

import threading
import time

def process_frame(frame_number):
    print(f"開始處理畫面 {frame_number}")
    time.sleep(1)  ## 模擬處理
    print(f"完成處理畫面 {frame_number}")

frame_numbers = range(1, 6)

threads = []
for frame in frame_numbers:
    thread = threading.Thread(target=process_frame, args=(frame,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有畫面處理完成")

執行結果說明

此方法可用於影片剪輯、特效處理等工作，透過每一格畫面平行處理，提高整體處理效率。

結論

多執行緒特別適合用於大量 I/O 操作或需要即時反應的應用場景。但對於高 CPU 負載的任務，仍須注意 GIL 的限制，適當選擇多進程處理將更有效率。

6. 使用多執行緒時的注意事項與最佳實踐

在 Python 中使用多執行緒雖能提升處理效率，但也容易遇到一些常見問題與潛在風險。本節將介紹多執行緒的挑戰，以及避免錯誤、確保穩定運作的最佳實踐。

注意事項

1. 全域直譯器鎖（GIL）的影響

Python 中的 GIL（Global Interpreter Lock）限制同一時間只能有一個執行緒執行 Python bytecode。這會使多執行緒在處理高 CPU 負載任務時，效能受限。

• 容易受到影響的情境：
• 大量的數學計算或密集運算處理
• 需要高度 CPU 使用率的演算法
• 解決方法：
• 使用 multiprocessing 模組，改用多進程處理
• 透過 C 擴充模組或 NumPy 等高效函式庫來避開 GIL 限制

2. 死結（Deadlock）

當多個執行緒彼此等待對方釋放資源時，會導致整個程式卡住，這種狀況稱為「死結」。

• 範例：
  執行緒 A 持有資源 X，等待資源 Y；同時執行緒 B 持有資源 Y，等待資源 X，造成互相等待。
• 解決方法：
• 統一所有執行緒的資源取得順序
• 使用 threading.RLock（可重入鎖）來避免死結

範例程式碼（避免死結）

import threading

lock1 = threading.Lock()
lock2 = threading.Lock()

def task1():
    with lock1:
        print("Task1 獲取 lock1")
        with lock2:
            print("Task1 獲取 lock2")

def task2():
    with lock2:
        print("Task2 獲取 lock2")
        with lock1:
            print("Task2 獲取 lock1")

thread1 = threading.Thread(target=task1)
thread2 = threading.Thread(target=task2)

thread1.start()
thread2.start()

thread1.join()
thread2.join()
print("兩個任務完成")

3. 賽局條件（Race Condition）

當多個執行緒同時讀寫同一資料，且操作順序不受控制時，可能導致資料不一致，這種情況稱為「賽局條件」。

• 範例：
  兩個執行緒同時對同一個計數器變數進行遞增操作時，可能導致最終結果錯誤。
• 解決方法：
• 使用 threading.Lock 來同步資料存取
• 盡量減少執行緒之間的共享資料

範例程式碼（使用鎖避免賽局條件）

import threading

lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:
        local_copy = counter
        local_copy += 1
        counter = local_copy

threads = [threading.Thread(target=increment) for _ in range(100)]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print(f"計數器結果: {counter}")

最佳實踐

1. 適當設定執行緒數量

• 設定執行緒數量時，應根據 CPU 核心數與 I/O 等待時間來調整。
• 建議：I/O 密集型任務可以使用較多執行緒；CPU 密集型任務建議控制在核心數以內。

2. 加入除錯與紀錄（Logging）

• 多執行緒程式難以除錯，妥善的紀錄對追蹤錯誤非常重要。
• 建議：使用 Python 的 logging 模組記錄每個執行緒的操作。

範例程式碼（使用 logging）

import threading
import logging

logging.basicConfig(level=logging.DEBUG, format='%(threadName)s: %(message)s')

def task():
    logging.debug("執行任務中")

threads = [threading.Thread(target=task) for _ in range(5)]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

logging.debug("所有任務完成")

3. 善用高階函式庫

使用 concurrent.futures.ThreadPoolExecutor 等高階工具，可簡化執行緒的管理與維護。

範例程式碼（ThreadPoolExecutor）

from concurrent.futures import ThreadPoolExecutor

def task(name):
    print(f"{name} 執行中")

with ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(task, ["任務1", "任務2", "任務3"])

結論

若要在 Python 中安全有效地使用多執行緒，需充分理解 GIL 的限制與執行緒間同步的重要性。透過適當的鎖定策略、除錯技巧，以及高階工具的輔助，可打造穩定可靠的並行處理架構。

7. 多執行緒與多進程的比較

在 Python 中實現平行處理時，常見的方式有「多執行緒」與「多進程」。兩者各有優勢與適用情境。本節將詳細比較兩者的差異，並提供選擇建議。

多執行緒與多進程的基本差異

詳細說明

• 多執行緒：
  因為所有執行緒在同一個進程中運作，啟動與資源使用上較輕量，資料共享也比較簡單。但在 Python 中，由於 GIL 的限制，在進行密集運算時效能可能會受限。
• 多進程：
  每個進程都有獨立記憶體與資源，因此可充分利用多核心 CPU，不受 GIL 限制。但若要進行資料交換，則需透過進程間通訊（IPC）機制，實作上較為複雜。

適合使用多執行緒的情況

• 適用範例：
• 網站爬蟲（Web Scraping）
• 檔案操作（讀取與寫入）
• 網路通訊（非同步處理）
• 原因：
  多執行緒適合處理有大量等待時間的 I/O 任務，能有效利用等待時間進行其他處理。同時，由於共用記憶體，資料傳遞也較簡單。

程式碼範例：I/O 為主的處理

import threading
import time

def file_operation(file_name):
    print(f"{file_name} 開始處理")
    time.sleep(2)  ## 模擬處理時間
    print(f"{file_name} 處理完成")

files = ["file1.txt", "file2.txt", "file3.txt"]

threads = []
for file in files:
    thread = threading.Thread(target=file_operation, args=(file,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有檔案處理完成")

適合使用多進程的情況

• 適用範例：
• 大型資料運算
• 機器學習模型訓練
• 影像處理、數值計算等 CPU 密集型任務
• 原因：
  多進程能繞過 GIL 限制，有效利用多核心 CPU，適合需要大量計算資源的情況。不過進程間的資料交換需特別處理。

程式碼範例：CPU 負載高的處理

from multiprocessing import Process
import time

def compute_heavy_task(task_id):
    print(f"任務 {task_id} 執行中")
    time.sleep(3)  ## 模擬計算
    print(f"任務 {task_id} 完成")

tasks = ["計算1", "計算2", "計算3"]

processes = []
for task in tasks:
    process = Process(target=compute_heavy_task, args=(task,))
    processes.append(process)
    process.start()

for process in processes:
    process.join()

print("所有計算任務已完成")

結合使用的情況

某些專案中，可將多執行緒與多進程結合使用，例如：使用多執行緒同時下載資料（I/O），下載完成後再使用多進程進行資料處理（CPU）。這樣能兼顧效率與效能。

選擇多執行緒與多進程的參考原則

選擇合適的平行處理方式時，可依下列幾點判斷：

1. 任務性質：

• 大量 I/O 等待：適合多執行緒
• 大量運算處理：適合多進程

1. 資源使用限制：

• 希望節省記憶體：多執行緒較適合
• 希望使用多核心 CPU：選擇多進程

1. 實作複雜度：

• 想簡單共享資料：使用多執行緒
• 能處理進程間通訊：可用多進程

8. 總結與常見問題（FAQ）

本篇文章針對 Python 中的多執行緒與多進程處理，從基本概念、實作範例、注意事項到使用時的判斷標準進行了完整說明。本節將統整重點，並以常見問答的形式補充說明，幫助讀者更深入理解。

本文重點整理

1. 多執行緒的特性

• 適合用於 I/O 等待時間較長的任務，資料共享簡單。
• 受到 GIL 限制，不適合大量計算任務。

1. 多進程的特性

• 不受 GIL 影響，適合進行高運算需求的處理。
• 記憶體空間獨立，資料傳遞需額外設計。

1. 選擇關鍵在於任務特性

• I/O 為主的任務 → 建議使用多執行緒
• CPU 為主的任務 → 建議使用多進程
• 必要時可結合兩者以達最佳效能

FAQ（常見問題）

Q1: 使用多執行緒時，建議的執行緒數量是多少？

A：
可依任務性質與系統資源決定。

• I/O 密集型：
  可啟動較多執行緒，根據同時處理的任務數調整。
• CPU 密集型：
  建議控制在 CPU 實體核心數以內，避免因 GIL 而導致效能反而下降。

Q2: 有辦法完全避開 GIL 的限制嗎？

A：
是的，有幾種方式可以減少或避免 GIL 的影響：

• 使用多進程：
  透過 multiprocessing 模組來平行處理，GIL 不會影響每個進程。
• 使用 C 擴充或高效函式庫：
  像是 NumPy、Pandas 等內部以 C 實作的函式庫，會在運算時暫時釋放 GIL。

Q3: 多執行緒與非同步處理（asyncio）有什麼不同？

A：

• 多執行緒：
  使用多個執行緒同時執行任務，可能會共用資源，需要處理同步與鎖。
• 非同步（asyncio）：
  在單一執行緒內以非同步方式處理任務，透過事件迴圈切換任務，適合大量 I/O 操作，效能高且避免競爭問題。

Q4: 使用執行緒池（ThreadPool）有什麼優點？

A：
使用執行緒池能有效管理大量任務，避免重複建立與銷毀執行緒所造成的資源浪費。concurrent.futures.ThreadPoolExecutor 是常用的高階工具。

範例：

from concurrent.futures import ThreadPoolExecutor

def task(name):
    print(f"{name} 執行中")

with ThreadPoolExecutor(max_workers=5) as executor:
    executor.map(task, ["任務1", "任務2", "任務3", "任務4", "任務5"])

Q5: 使用多執行緒會讓記憶體使用量增加嗎？

A：
多執行緒會共用主程式的記憶體，因此單純增加執行緒數量並不會大量增加記憶體使用量。但每個執行緒仍會分配一定的堆疊記憶體，如果啟動太多執行緒，總體記憶體使用量仍可能上升。

結語

多執行緒與多進程是提升 Python 程式效能的重要手段。根據任務特性、系統資源與實作需求靈活運用，將能有效提升處理效率與應用程式的反應速度。希望本篇文章能幫助你建立平行處理的概念與實作能力。