目錄
04-2. Race Condition（競態條件）
⏱️ 閱讀時間： 12 分鐘 🎯 難度： ⭐⭐⭐ (中等)
🎯 本篇重點
深入理解 Race Condition 的成因、識別方法、影響，以及如何預防和解決這個最常見的 Thread Safety 問題。
🤔 什麼是 Race Condition？
Race Condition（競態條件） = 多個 Thread「競賽」存取共享資源，結果取決於執行順序
一句話解釋： 當程式的執行結果取決於 Thread 的執行順序或時機，而這個順序是不可預測的，就發生了 Race Condition。
🏃 用賽跑來比喻
正常的賽跑（無共享資源）
跑者 A ────────→ 終點 A
跑者 B ────────→ 終點 B
各自跑各自的道，互不干擾Race Condition（共享資源）
跑者 A ─┐
├──→ 同一個計分板 ← 誰先到誰寫入
跑者 B ─┘
問題：
- A 看到分數 = 0，準備寫入 1
- B 看到分數 = 0，準備寫入 1
- A 寫入 1
- B 寫入 1（覆蓋了 A 的結果）
- 最終分數 = 1（應該是 2！）💻 經典 Race Condition 案例
案例 1：計數器問題
from threading import Thread
import time
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、180000 等（每次不同）執行 10 次的結果：
Run 1: 156234
Run 2: 178901
Run 3: 164567
Run 4: 192345
Run 5: 143210
Run 6: 187654
Run 7: 159876
Run 8: 175432
Run 9: 168901
Run 10: 181234
每次都不同！這就是 Race Condition案例 2：銀行轉帳
from threading import Thread
import time
class BankAccount:
def __init__(self, balance):
self.balance = balance
def withdraw(self, amount):
"""不安全的提款"""
# 1. 檢查餘額
if self.balance >= amount:
print(f"檢查通過，餘額: {self.balance}")
# 模擬處理時間（問題發生處）
time.sleep(0.001)
# 2. 提款
self.balance -= amount
print(f"提款 {amount}，剩餘: {self.balance}")
else:
print("餘額不足")
def concurrent_withdraw(account, amount):
account.withdraw(amount)
# 帳戶餘額 1000
account = BankAccount(1000)
# 兩個 Thread 同時提款 600
t1 = Thread(target=concurrent_withdraw, args=(account, 600))
t2 = Thread(target=concurrent_withdraw, args=(account, 600))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"最終餘額: {account.balance}")可能的輸出：
檢查通過，餘額: 1000 ← Thread 1
檢查通過，餘額: 1000 ← Thread 2（也通過了！）
提款 600，剩餘: 400
提款 600，剩餘: -200 ← 變成負數！
最終餘額: -200案例 3：Check-Then-Act 問題
from threading import Thread
class Cache:
def __init__(self):
self.data = {}
def get_or_create(self, key):
"""不安全的 get or create"""
# Check
if key not in self.data:
print(f"創建 {key}")
# 模擬耗時操作
import time
time.sleep(0.01)
# Act
self.data[key] = f"Value for {key}"
return self.data[key]
cache = Cache()
def access_cache(key):
result = cache.get_or_create(key)
print(f"獲得: {result}")
# 多個 Thread 同時訪問同一個 key
threads = [Thread(target=access_cache, args=('user_123',)) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()可能的輸出：
創建 user_123
創建 user_123 ← 重複創建！
創建 user_123
創建 user_123
創建 user_123
獲得: Value for user_123
獲得: Value for user_123
...🔍 Race Condition 的類型
1. Read-Modify-Write
# 經典的三步驟 Race
counter = 0
def unsafe_increment():
global counter
# 1. Read（讀取）
temp = counter
# 2. Modify（修改）
temp = temp + 1
# 3. Write（寫入）
counter = temp
# Thread A: Read(0) → Modify(1) → Write(1)
# Thread B: Read(0) → Modify(1) → Write(1)
# 結果：1（應該是 2）2. Check-Then-Act
# 檢查後執行的 Race
balance = 1000
def unsafe_withdraw(amount):
global balance
# Check（檢查）
if balance >= amount:
# 其他 Thread 可能在這裡插入
# Act（執行）
balance -= amount
# Thread A: Check(通過) → Act(提款)
# Thread B: Check(通過) → Act(提款)
# 可能導致餘額不足3. Lazy Initialization
# 延遲初始化的 Race
instance = None
def get_instance():
global instance
if instance is None: # Check
instance = ExpensiveObject() # Act
return instance
# 多個 Thread 可能創建多個 instance🛡️ 解決 Race Condition
解決方案 1：使用 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（正確）解決方案 2：銀行轉帳修正
from threading import Thread, Lock
import time
class SafeBankAccount:
def __init__(self, balance):
self.balance = balance
self.lock = Lock()
def withdraw(self, amount):
"""安全的提款"""
with self.lock: # 整個操作原子化
if self.balance >= amount:
print(f"檢查通過，餘額: {self.balance}")
time.sleep(0.001)
self.balance -= amount
print(f"提款 {amount}，剩餘: {self.balance}")
return True
else:
print("餘額不足")
return False
account = SafeBankAccount(1000)
threads = [
Thread(target=lambda: account.withdraw(600)),
Thread(target=lambda: account.withdraw(600))
]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"最終餘額: {account.balance}")輸出：
檢查通過，餘額: 1000
提款 600，剩餘: 400
餘額不足 ← 正確阻止了
最終餘額: 400解決方案 3：雙重檢查鎖定（Double-Checked Locking）
from threading import Lock
class Singleton:
_instance = None
_lock = Lock()
@classmethod
def get_instance(cls):
# 第一次檢查（無鎖，快）
if cls._instance is None:
with cls._lock:
# 第二次檢查（有鎖，安全）
if cls._instance is None:
print("創建實例")
cls._instance = Singleton()
return cls._instance
# 測試
from threading import Thread
def get_singleton():
instance = Singleton.get_instance()
print(f"獲得實例: {id(instance)}")
threads = [Thread(target=get_singleton) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()輸出：
創建實例 ← 只創建一次
獲得實例: 140234567890123
獲得實例: 140234567890123
...（所有 ID 相同）🧪 檢測 Race Condition
方法 1：壓力測試
from threading import Thread
import time
def test_race_condition(func, num_threads=10, iterations=100):
"""檢測 Race Condition"""
results = []
for _ in range(iterations):
# 重置狀態
global counter
counter = 0
# 啟動多個 Thread
threads = [Thread(target=func) for _ in range(num_threads)]
for t in threads:
t.start()
for t in threads:
t.join()
results.append(counter)
# 分析結果
expected = num_threads * 100000
unique_results = set(results)
print(f"預期結果: {expected}")
print(f"實際結果數量: {len(unique_results)}")
print(f"最小值: {min(results)}")
print(f"最大值: {max(results)}")
print(f"正確次數: {results.count(expected)}/{iterations}")
if len(unique_results) > 1:
print("⚠️ 檢測到 Race Condition！")
else:
print("✅ 未檢測到 Race Condition")
# 測試不安全的版本
def unsafe_increment():
global counter
for _ in range(100000):
counter += 1
print("=== 測試不安全版本 ===")
test_race_condition(unsafe_increment, num_threads=2, iterations=10)方法 2：使用工具
# 使用 Python 的 threading 內建工具
import threading
def check_current_thread():
"""檢查當前 Thread 資訊"""
thread = threading.current_thread()
print(f"Thread: {thread.name}")
print(f"Thread ID: {thread.ident}")
print(f"Is alive: {thread.is_alive()}")
print(f"Active threads: {threading.active_count()}")
print(f"All threads: {threading.enumerate()}")🎯 預防 Race Condition 的最佳實踐
1. 最小化共享狀態
# ❌ 不好：共享全域變數
counter = 0
def increment():
global counter
counter += 1
# ✅ 好：避免共享
from queue import Queue
def worker(task_queue, result_queue):
"""每個 Thread 獨立工作"""
while not task_queue.empty():
task = task_queue.get()
result = process(task)
result_queue.put(result)2. 使用 Thread-Safe 資料結構
from queue import Queue
from threading import Thread
# Queue 內建 Thread Safety
task_queue = Queue()
result_queue = Queue()
def producer():
for i in range(10):
task_queue.put(i) # 安全
def consumer():
while True:
try:
task = task_queue.get(timeout=1) # 安全
result = task * 2
result_queue.put(result)
except:
break3. 原子操作
from threading import Thread
import threading
# Python 的某些操作是原子的
x = 0
x = 1 # ✅ 原子（單一賦值）
# 但這些不是原子的
x += 1 # ❌ 非原子（read-modify-write）
x = x + 1 # ❌ 非原子4. 不可變物件
from threading import Thread
# ✅ 使用不可變物件（Immutable）
from collections import namedtuple
Config = namedtuple('Config', ['host', 'port'])
config = Config('localhost', 8080)
def worker():
# 只讀取，不修改
print(f"連線到 {config.host}:{config.port}")
# 安全，因為 config 不可變📊 Race Condition 影響評估
嚴重程度分級
| 等級 | 影響 | 範例 |
|---|---|---|
| 🔴 嚴重 | 資料損壞、金額錯誤 | 銀行轉帳、訂單處理 |
| 🟡 中等 | 資料不一致、重複操作 | 快取更新、日誌記錄 |
| 🟢 輕微 | 效能影響、順序錯亂 | 統計計數、日誌順序 |
實際案例：Therac-25 事件
1985-1987 年，Therac-25 放射治療機器
因為 Race Condition 導致：
- 過量輻射照射患者
- 至少 6 人受傷或死亡
- 原因：軟體中的 Race Condition
教訓：Thread Safety 關乎人命！✅ 重點回顧
Race Condition 定義：
- 程式結果取決於 Thread 執行順序
- 執行順序不可預測
- 導致不一致的結果
常見類型：
- Read-Modify-Write - 讀取、修改、寫入
- Check-Then-Act - 檢查後執行
- Lazy Initialization - 延遲初始化
解決方案：
- ✅ 使用 Lock 保護臨界區
- ✅ 雙重檢查鎖定（DCL）
- ✅ 使用 Thread-Safe 資料結構
- ✅ 避免共享狀態
檢測方法：
- ✅ 壓力測試（多次執行）
- ✅ 檢查結果一致性
- ✅ 使用工具監控
預防原則：
- ✅ 最小化共享狀態
- ✅ 使用不可變物件
- ✅ 優先使用 Thread-Safe API
- ✅ 設計階段就考慮 Thread Safety
關鍵理解：
- ⚠️ Race Condition 難以重現和除錯
- ⚠️ 測試通過不代表沒有問題
- ⚠️ 必須從設計上避免
- ⚠️ 影響可能非常嚴重
上一篇： 04-1. Thread Safety 基礎概念 下一篇： 04-3. Deadlock（死鎖）
最後更新：2025-01-06