Python 多執行緒入門：thread、threading 與 Queue 模組實戰

前言

本文整理 Python 多執行緒核心模組 thread（低階）、threading（高階）與 queue（執行緒安全佇列）的用法，並提供完整範例說明啟動、同步與資料傳遞。

核心模組比較

範例一：thread.start_new_thread 基礎用法

import thread, time

def worker(no, interval):
    for i in range(5):
        print(f"Thread-{no} sleep {interval}s, time: {time.ctime()}")
        time.sleep(interval)
    print(f"Thread-{no} finished")

thread.start_new_thread(worker, (1, 2))
thread.start_new_thread(worker, (2, 1))
time.sleep(10)  # 主執行緒等待

範例二：threading.Thread 物件導向寫法

import threading, time, random

locks = []
for _ in range(2):
    lock = threading.Lock()
    lock.acquire()
    locks.append(lock)

def task(n, sec, lock):
    print(f"Thread {n} start, sleep {sec}s")
    time.sleep(sec)
    print(f"Thread {n} end")
    lock.release()

threads = []
for i, lock in enumerate(locks):
    t = threading.Thread(target=task, args=(i, random.choice([2,4]), lock))
    t.start()
    threads.append(t)

for lock in locks:
    while lock.locked(): pass  # 等待釋放

print("All threads done")

範例三：Queue 實作生產者-消費者

import threading, queue, time, random

q = queue.Queue(maxsize=5)

def producer():
    for i in range(10):
        q.put(i)
        print(f"Produced {i}")
        time.sleep(random.random())

def consumer():
    while True:
        item = q.get()
        print(f"Consumed {item}")
        q.task_done()
        time.sleep(random.random())

threading.Thread(target=producer).start()
threading.Thread(target=consumer, daemon=True).start()
q.join()
print("Done")

注意事項

• 主執行緒結束會強制終止所有非守護執行緒，務必使用 join() 或鎖機制等待。
• thread 模組在 Python 3 中改名為 _thread，建議直接使用 threading。
• 共享資料需加鎖或使用 queue.Queue 避免競爭條件。