目錄
04-1. Thread Safety 基礎概念
⏱️ 閱讀時間： 10 分鐘 🎯 難度： ⭐⭐ (簡單)
🤔 什麼是 Thread Safety？
一句話解釋： Thread Safety（線程安全）是指多個 Thread 同時存取共享資源時，程式仍能正確執行，不會產生錯誤結果。
🏢 用銀行帳戶來比喻
不安全的情況
銀行帳戶：餘額 = 1000 元
時間 T0:
├─ 員工 A（Thread A）: 讀取餘額 = 1000
└─ 員工 B（Thread B）: 讀取餘額 = 1000
時間 T1:
├─ 員工 A: 存入 100 → 計算 1000 + 100 = 1100
└─ 員工 B: 提取 50 → 計算 1000 - 50 = 950
時間 T2:
├─ 員工 A: 寫入 1100
└─ 員工 B: 寫入 950
最終餘額: 950 ← 錯誤！應該是 1050這就是 Thread Safety 問題！
🔍 實際案例
案例 1：計數器問題
from threading import Thread
counter = 0
def increment():
global counter
for _ in range(100000):
counter += 1 # 這不是原子操作！
# 創建兩個 Thread
t1 = Thread(target=increment)
t2 = Thread(target=increment)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"Counter: {counter}")
# 預期: 200000
# 實際: 可能是 150000 或其他值 ← 不安全！為什麼會錯誤？
# counter += 1 實際上是三個步驟：
1. temp = counter # 讀取
2. temp = temp + 1 # 計算
3. counter = temp # 寫入
# 兩個 Thread 可能交錯執行：
Thread A: temp_a = counter (100)
Thread B: temp_b = counter (100) ← 還是 100！
Thread A: temp_a = 101
Thread B: temp_b = 101
Thread A: counter = 101
Thread B: counter = 101 ← 覆蓋了 A 的結果！案例 2：使用 Lock 解決
from threading import Thread, Lock
counter = 0
lock = Lock()
def safe_increment():
global counter
for _ in range(100000):
with lock: # 加鎖
counter += 1
t1 = Thread(target=safe_increment)
t2 = Thread(target=safe_increment)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"Counter: {counter}")
# 輸出: 200000 ← 正確！Lock 的作用：
Thread A 獲得 Lock:
counter += 1 ← 只有 A 可以執行
Thread A 釋放 Lock
Thread B 獲得 Lock:
counter += 1 ← 現在 B 可以執行
Thread B 釋放 Lock
確保同一時間只有一個 Thread 修改 counter！📋 Thread Safety 的三大問題
1. Race Condition（競態條件）
定義： 多個 Thread 競爭存取共享資源，結果取決於執行順序。
# ❌ 不安全
balance = 1000
def withdraw(amount):
global balance
if balance >= amount: # 檢查
# 其他 Thread 可能在這裡插入！
balance -= amount # 提取
# 兩個 Thread 同時提取 600
# 可能都通過檢查，導致餘額變負數！2. Deadlock（死鎖）
定義： 多個 Thread 互相等待對方釋放資源，導致永遠阻塞。
from threading import Lock
lock_a = Lock()
lock_b = Lock()
def task1():
with lock_a:
print("Task1 獲得 Lock A")
with lock_b: # 等待 Lock B
print("Task1 獲得 Lock B")
def task2():
with lock_b:
print("Task2 獲得 Lock B")
with lock_a: # 等待 Lock A
print("Task2 獲得 Lock A")
# Task1 持有 A，等待 B
# Task2 持有 B，等待 A
# 互相等待 → Deadlock！3. Data Race（資料競爭）
定義： 多個 Thread 同時存取同一記憶體位址，且至少有一個是寫入操作。
# ❌ Data Race
shared_data = []
def writer():
shared_data.append(1) # 寫入
def reader():
if len(shared_data) > 0: # 讀取
print(shared_data[0])
# Writer 和 Reader 同時執行
# 可能讀到不一致的狀態！🔧 確保 Thread Safety 的方法
1. Lock（互斥鎖）
from threading import Lock
lock = Lock()
def critical_section():
with lock:
# 臨界區：同一時間只有一個 Thread
shared_resource.modify()2. RLock（可重入鎖）
from threading import RLock
rlock = RLock()
def recursive_function(n):
with rlock:
if n > 0:
recursive_function(n - 1) # 可以重複獲得鎖3. Semaphore（信號量）
from threading import Semaphore
# 最多允許 3 個 Thread 同時執行
semaphore = Semaphore(3)
def limited_access():
with semaphore:
# 最多 3 個 Thread 可以同時進入
use_resource()4. Queue（線程安全佇列）
from queue import Queue
# Queue 內建 Thread Safety
q = Queue()
def producer():
q.put(data) # 安全
def consumer():
data = q.get() # 安全🎯 Thread Safety 檢查清單
設計階段
- 識別所有共享變數
- 確定哪些操作需要同步
- 選擇合適的同步機制
實作階段
- 使用 Lock 保護臨界區
- 避免嵌套鎖（防止 Deadlock）
- 使用線程安全的資料結構
測試階段
- 壓力測試（大量並發）
- 檢查 Race Condition
- 監控 Deadlock
✅ 重點回顧
Thread Safety 的定義：
- 多 Thread 同時存取共享資源時，程式仍能正確執行
三大問題：
- Race Condition：執行順序影響結果
- Deadlock：互相等待資源
- Data Race：同時讀寫同一記憶體
解決方法：
- ✅ Lock（互斥鎖）
- ✅ RLock（可重入鎖）
- ✅ Semaphore（信號量）
- ✅ 使用線程安全的資料結構
關鍵理解：
- ✅ 共享記憶體是問題來源
- ✅ 同步機制確保安全
- ✅ 過度同步影響效能
- ✅ 設計階段就要考慮 Thread Safety
下一篇： 04-2. Race Condition（競態條件）
最後更新：2025-01-04