Python 執行緒(Threading)
在多工處理（Multitasking）的現代軟體開發中，「執行緒」是提升程式回應效率的關鍵技術。本文將從作業系統底層架構出發，逐步過渡到 Python 的實作層面。
一、 核心觀念：程序 vs 執行緒與系統資源的關係
在進入程式碼之前，必須先理清作業系統（OS）如何管理執行中的軟體。
1. 什麼是程序 (Process)？
定義：在作業系統中，一個「執行中的程式」就被稱為一個程序。例如當你在 Windows 開啟 Google Chrome 或 Line，系統就會在主記憶體（RAM）中為它們分派獨立、隔離的記憶體空間。
特性：程序之間是各自獨立的，一個程序崩潰（如 Chrome 某個分頁當掉）通常不會直接導致另一個程序（如 Line）當掉。
2. 什麼是執行緒 (Thread)？
定義：執行緒是附屬於程序之下的微型執行單位，它是 CPU 進行工作排程（Scheduling）的最小核心單元。
特性：同一個程序內的多個執行緒會共享該程序的記憶體空間（RAM）。例如一個 Google Chrome 程序內部可能同時跑了 20 個執行緒，分別處理網頁渲染、網路下載、JavaScript 解析等。
3. CPU Core、RAM、Process、Thread 的階層關係
當現代 CPU 擁有複數核心（Multi-core）時，作業系統會將程序分配給特定的核心去執行。
核心 1 (Core 1) $\rightarrow$ 程序 1 (Process 1) 在獨立的 RAM 1 內運行 $\rightarrow$ 內部切分出 執行緒 A、執行緒 B。
同一個程序內的執行緒（Thread A 與 B）雖然看起來像在「同時」運作，但若是使用 Python，受限於其特殊的 GIL（全域解釋器鎖, Global Interpreter Lock） 機制，在同一個時間點，實際上只有一個執行緒能在同一個核心內被真正執行。作業系統是透過極為快速的「上下文切換（Context Switching）」，才讓人類體感上覺得它們在並行運作。
而這些程序是在作業系統裡執行中的程式，且在ram上有自己的記憶體空間，從上圖可以知道電腦裡有很多的程序在同一時間運行。而執行緒(thread)是附屬於程序的，一個程序可以擁有很多個執行緒，例如上圖中的Google Chrome可能有20個執行緒在運行中。現在的CPU有多個核心(Core)數，每一個程序會被作業系統分配到各自的 CPU Core裡。CPU Core、RAM、Process、Thread的關係如下圖：
二、 實作：Python 執行緒的建立與執行（底層 _thread 模組）
Python 提供兩種方式來操作執行緒：低階的 _thread（舊稱 thread）與高階的 threading 模組。本文採用了底層的 _thread 進行示範：
1. 核心函數語法
_thread.start_new_thread( function, args[, kwargs] )
這個函數會立即配置一個新的執行緒，並在該執行緒中呼叫傳入的 function。
function：執行緒要執行的函數名稱。args：傳遞給函數的參數，必須是元組 (Tuple) 格式。
2. 範例程式碼逐行拆解
import time
t = 0
# 1. 定義子執行緒要執行的任務函數
def prt_time(threadID, interval):
cnt = 0
while cnt < 3:
time.sleep(interval) # 讓該執行緒暫停（睡覺）指定秒數
cnt += 1
# 輸出當前執行緒的 ID 與系統時間
print("%s 現在時間:%s" % (threadID, time.ctime(time.time())))
# 2. 嘗試建立並啟動兩個子執行緒
try:
# 建立執行緒 A：每隔 1 秒印出一次時間，共印 3 次
_thread.start_new_thread(prt_time, ("執行緒A", 1, ))
# 建立執行緒 B：每隔 3 秒印出一次時間，共印 3 次
_thread.start_new_thread(prt_time, ("執行緒B", 3, ))
except:
print("錯誤: 執行緒無法執行")
# 3. 主執行緒（Main Thread）的生命週期維護
while t <= 10:
time.sleep(1.5) # 主執行緒每隔 1.5 秒醒來一次
print("主執行緒現在時間:%s" % (time.ctime(time.time())))
t = t + 1
print("主執行緒結束")
🔍 關鍵邏輯與細節說明
Tuple 參數的陷阱：在
("執行緒A", 1, )中，最後面的逗號,絕對不能省略。因為在 Python 中，單一元素的括號("執行緒A")只會被當成字串，加上逗號才會被正確識別為 Tuple。為什麼主執行緒需要
while t <= 10控制？ 在作業系統的架構中，一旦「主執行緒（Main Thread）」執行結束，整個程序（Process）就會被系統強制關閉，此時附屬在底下的所有子執行緒不論有沒有跑完都會被瞬間掐斷。因此，程式末端設計了一個跑 10 次、每次睡 1.5 秒的迴圈（共維持約 15 秒），目的是「確保主執行緒活得夠久，讓子執行緒 A 與 B 有足夠的時間把任務執行完畢」。
範例程式結果：
三、 技術補充：給小編的擴充建議（進階的 threading 模組）
引入 Python 現代標準庫所推薦的 threading 模組。相較於原有的 _thread，高階的 threading 提供了更好維護的物件導向設計，以及解決主執行緒過早結束的 join() 機制：
💡 擴充範例（更優雅的寫法）：
充範例（更優雅的寫法）：
import time
def prt_time(name, delay):
for i in range(3):
time.sleep(delay)
print(f"{name} 現在時間: {time.ctime()}")
# 使用 threading.Thread 建立物件
threadA = threading.Thread(target=prt_time, args=("執行緒A", 1))
threadB = threading.Thread(target=prt_time, args=("執行緒B", 3))
# 啟動執行緒
threadA.start()
threadB.start()
# 關鍵：使用 join() 讓主執行緒「原地等待」子執行緒完成，不需要再寫死 while 迴圈！
threadA.join()
threadB.join()
print("主執行緒在子執行緒都結束後，才安全地劃下句點。")
四、執行緒同步化
有時執行緒必須等待另一個執行緒結束才能執行。這時可以使用 lock 的機制。範例程式：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | import threading import time t = 0 lock = threading.Lock() class myThread (threading.Thread): def __init__(self, threadID, delay): threading.Thread.__init__(self) self.id = threadID self.delay = delay def run(self): print ("執行 " + self.id) lock.acquire() prt_time(self.id, self.delay) lock.release() print ("離開 " + self.id) # 執行緒所用的輸出系統時間之函數 def prt_time( threadID, delay): cnt = 0 while cnt < 3: time.sleep(delay) cnt += 1 print( "%s 現在時間:%s" % ( threadID, time.ctime(time.time()) ) ) # 建立兩個執行緒 thread1 = myThread("執行緒A", 1) thread2 = myThread("執行緒B", 3) # 啟動執行緒 thread1.start() thread2.start() # 等待執行緒結束 thread1.join() thread2.join() print ("離開主執行緒") |
範例程式結果：
📢小編碎碎念：
多執行緒在處理 I/O 密集型工作（如網路爬蟲、檔案讀寫）時能大幅縮短等待時間！大家在寫 Python 多執行緒時，有遇過資料搶奪（Race Condition）的問題嗎？
若您覺得文章寫得不錯，請點選文章上的廣告，來支持小編，謝謝。
If you like this post, please click the ads on the blog or buy me a coffee. Thank you very much.
留言