資料庫交易與並發控制：Django 開發者必修課
從理論到實戰，徹底理解並發問題與解決方案
目錄
為什麼並發控制如此重要？
想像一下黑色星期五的購物網站：成千上萬的用戶同時搶購限量商品。如果沒有適當的並發控制，可能會發生超賣、訂單遺失，甚至帳戶餘額錯誤等災難性後果。
這就是為什麼每個後端開發者都必須深入理解交易與並發控制。
📚 交易基礎：全有或全無
什麼是交易（Transaction）？
交易是資料庫中一組不可分割的操作序列，要麼全部成功執行，要麼全部不執行。
生活中的交易
ATM 轉帳的例子：
當你從 A 帳戶轉 1000 元到 B 帳戶時，實際上包含兩個步驟：
- A 帳戶扣款 1000 元
- B 帳戶增加 1000 元
如果第一步成功但第二步失敗，錢就憑空消失了！交易機制確保：
- 兩步都成功 → 轉帳完成
- 任一步失敗 → 全部回滾，錢還在 A 帳戶
交易的 ACID 特性
- Atomicity（原子性）：不可分割，全有或全無
- Consistency（一致性）：始終保持資料的有效狀態
- Isolation（隔離性）：多個交易互不干擾
- Durability（持久性）：提交後永久保存
👻 三大異常現象：並發的幽靈
當多個交易同時執行時，如果沒有適當的控制，就會出現以下異常現象：
1. 髒讀（Dirty Read）：讀到未提交資料
購物車的例子：
時間線：
10:00 - 用戶A：將商品價格從 100 改為 50（未提交）
10:01 - 用戶B：看到價格 50，高興地加入購物車
10:02 - 用戶A：取消修改（回滾）
10:03 - 用戶B：結帳時發現價格變回 100，錯愕！用戶 B 讀到了用戶 A 未提交的「髒」資料。
2. 不可重複讀（Non-repeatable Read）：同筆資料兩次查詢值不同
查看餘額的例子：
時間線：
10:00 - 你查詢帳戶餘額：5000 元
10:01 - 你思考要買什麼
10:02 - 配偶從同帳戶提領 2000 元
10:03 - 你再次確認餘額：3000 元（怎麼變少了？）在同一個交易中，兩次讀取同一筆資料得到不同結果。
3. 幻讀（Phantom Read）：查詢結果筆數改變
統計訂單的例子：
時間線：
10:00 - 管理員查詢今日訂單數：100 筆
10:01 - 新用戶下單 5 筆
10:02 - 管理員準備匯出 100 筆訂單
10:03 - 實際匯出 105 筆（多出幽靈訂單！）查詢條件相同，但結果集的大小改變了。
🛡️ 隔離級別：控制異常的防護罩
資料庫提供不同的隔離級別來控制這些異常，從寬鬆到嚴格分為四級：
1. Read Uncommitted（讀取未提交）
- 特點：最寬鬆，可能發生所有異常
- 性能：最快
- 使用場景：幾乎不使用
2. Read Committed（讀取已提交）
- 特點：防止髒讀
- 可能發生：不可重複讀、幻讀
- 性能：較快
- 使用場景：PostgreSQL 預設級別
3. Repeatable Read（可重複讀）
- 特點：防止髒讀、不可重複讀
- 可能發生：幻讀（MySQL InnoDB 實際上也防止了）
- 性能：中等
- 使用場景：MySQL 預設級別
4. Serializable（可序列化）
- 特點：防止所有異常，如同依序執行
- 性能：最慢
- 使用場景：銀行等高度敏感系統
隔離級別對照表
| 隔離級別 | 髒讀 | 不可重複讀 | 幻讀 | 性能 |
|---|---|---|---|---|
| Read Uncommitted | 可能 | 可能 | 可能 | 最快 |
| Read Committed | 不可能 | 可能 | 可能 | 快 |
| Repeatable Read | 不可能 | 不可能 | 可能* | 中等 |
| Serializable | 不可能 | 不可能 | 不可能 | 最慢 |
*註：MySQL InnoDB 透過 Next-Key Locking 實際上也防止了幻讀
💡 Django 解決方案
Django 提供了多種工具來處理交易和並發問題：
1. transaction.atomic()：保證原子性
確保一組操作要麼全部成功，要麼全部失敗。
裝飾器用法：
from django.db import transaction
@transaction.atomic
def transfer_money(from_account_id, to_account_id, amount):
from_account = Account.objects.get(id=from_account_id)
to_account = Account.objects.get(id=to_account_id)
from_account.balance -= amount
from_account.save()
to_account.balance += amount
to_account.save()
# 如果任何步驟失敗，整個交易會回滾上下文管理器用法：
def create_order(user, items):
try:
with transaction.atomic():
# 創建訂單
order = Order.objects.create(user=user, total=0)
total = 0
for item in items:
# 檢查庫存
product = Product.objects.get(id=item['product_id'])
if product.stock < item['quantity']:
raise InsufficientStockError(f"{product.name} 庫存不足")
# 創建訂單項目
OrderItem.objects.create(
order=order,
product=product,
quantity=item['quantity'],
price=product.price
)
# 扣減庫存
product.stock -= item['quantity']
product.save()
total += product.price * item['quantity']
# 更新訂單總額
order.total = total
order.save()
except InsufficientStockError:
# 交易會自動回滾
raise2. select_for_update()：悲觀鎖防止並發修改
在查詢時鎖定資料列，防止其他交易修改。
基本用法：
from django.db import transaction
@transaction.atomic
def withdraw_money(account_id, amount):
# 鎖定帳戶，其他交易必須等待
account = Account.objects.select_for_update().get(id=account_id)
if account.balance < amount:
raise InsufficientFundsError("餘額不足")
account.balance -= amount
account.save()
return account.balance處理死鎖：
@transaction.atomic
def transfer_with_deadlock_prevention(from_id, to_id, amount):
# 始終按照 ID 順序鎖定，避免死鎖
if from_id < to_id:
account1 = Account.objects.select_for_update().get(id=from_id)
account2 = Account.objects.select_for_update().get(id=to_id)
from_account, to_account = account1, account2
else:
account1 = Account.objects.select_for_update().get(id=to_id)
account2 = Account.objects.select_for_update().get(id=from_id)
from_account, to_account = account2, account1
# 執行轉帳
from_account.balance -= amount
to_account.balance += amount
from_account.save()
to_account.save()3. F() 表達式：避免競爭條件
直接在資料庫層級執行更新，避免 Read-Modify-Write 競爭。
問題範例（有競爭條件）：
# 危險！可能造成遺失更新
product = Product.objects.get(id=1)
product.view_count += 1 # 讀取、修改
product.save() # 寫入
# 如果兩個請求同時執行：
# 請求1：讀取 view_count = 100
# 請求2：讀取 view_count = 100
# 請求1：設定 view_count = 101
# 請求2：設定 view_count = 101
# 結果：只增加了 1 次，而非 2 次！解決方案（使用 F() 表達式）：
from django.db.models import F
# 安全！在資料庫層級原子性更新
Product.objects.filter(id=1).update(
view_count=F('view_count') + 1
)
# 更複雜的例子：扣減庫存
from django.db import transaction
@transaction.atomic
def purchase_product(product_id, quantity):
# 使用 F() 表達式原子性扣減庫存
updated = Product.objects.filter(
id=product_id,
stock__gte=quantity # 確保庫存足夠
).update(
stock=F('stock') - quantity
)
if not updated:
raise InsufficientStockError("庫存不足或商品不存在")
# 如果需要返回更新後的值
product = Product.objects.get(id=product_id)
return product🎯 實戰：組合使用
單獨使用 atomic() 往往不夠，需要結合其他機制：
秒殺系統範例
from django.db import transaction
from django.db.models import F
from django.core.cache import cache
import uuid
class SeckillService:
@staticmethod
@transaction.atomic
def create_order(user_id, product_id, quantity=1):
# 1. 檢查用戶是否已購買（使用快取加速）
cache_key = f"seckill:{product_id}:user:{user_id}"
if cache.get(cache_key):
raise AlreadyPurchasedError("您已經購買過此商品")
# 2. 使用 select_for_update 鎖定商品
try:
product = Product.objects.select_for_update(
nowait=True # 不等待，立即失敗
).get(id=product_id, is_seckill=True)
except Product.DoesNotExist:
raise ProductNotFoundError("商品不存在")
except DatabaseError:
# nowait=True 時，如果無法立即獲得鎖會拋出異常
raise SystemBusyError("系統繁忙，請稍後再試")
# 3. 檢查庫存
if product.stock < quantity:
raise InsufficientStockError("商品已售罄")
# 4. 使用 F() 表達式扣減庫存
Product.objects.filter(id=product_id).update(
stock=F('stock') - quantity,
sold_count=F('sold_count') + quantity
)
# 5. 創建訂單
order = Order.objects.create(
order_no=str(uuid.uuid4()),
user_id=user_id,
product=product,
quantity=quantity,
total_amount=product.seckill_price * quantity,
status='pending_payment'
)
# 6. 標記用戶已購買
cache.set(cache_key, True, 3600) # 快取 1 小時
return order庫存管理最佳實踐
class InventoryService:
@staticmethod
def adjust_stock(product_id, quantity_change, reason):
"""
調整庫存的統一入口
quantity_change: 正數表示入庫，負數表示出庫
"""
with transaction.atomic():
# 鎖定產品
product = Product.objects.select_for_update().get(id=product_id)
# 計算新庫存
new_stock = product.stock + quantity_change
# 驗證庫存不能為負
if new_stock < 0:
raise InsufficientStockError(
f"庫存不足，當前庫存：{product.stock}，"
f"嘗試扣減：{abs(quantity_change)}"
)
# 使用 F() 表達式更新，確保原子性
Product.objects.filter(id=product_id).update(
stock=F('stock') + quantity_change,
updated_at=timezone.now()
)
# 記錄庫存變動歷史
StockHistory.objects.create(
product=product,
change_quantity=quantity_change,
after_quantity=new_stock,
reason=reason,
created_at=timezone.now()
)
return new_stock🚨 常見陷阱與解決方案
陷阱 1：長交易
問題：交易時間過長會鎖定資源，影響並發性能。
解決方案：
# 錯誤示範
@transaction.atomic
def process_large_order(items):
for item in items: # 假設有 1000 個項目
# 複雜的業務邏輯
validate_item(item)
calculate_discount(item)
update_inventory(item)
send_notification(item) # 耗時的外部調用！
# 正確做法
def process_large_order(items):
# 1. 先在交易外準備資料
prepared_items = []
for item in items:
validated = validate_item(item)
discount = calculate_discount(item)
prepared_items.append({
'item': item,
'validated': validated,
'discount': discount
})
# 2. 只在交易中執行資料庫操作
with transaction.atomic():
for prepared in prepared_items:
update_inventory(prepared['item'], prepared['discount'])
# 3. 交易完成後執行耗時操作
for item in items:
send_notification.delay(item) # 使用 Celery 異步執行陷阱 2：忽視資料庫預設隔離級別
問題：不同資料庫的預設隔離級別不同。
解決方案：
# settings.py
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
# PostgreSQL 預設是 READ COMMITTED
'OPTIONS': {
'isolation_level': psycopg2.extensions.ISOLATION_LEVEL_REPEATABLE_READ,
},
}
}
# 或在需要時動態設定
from django.db import connection
with transaction.atomic():
cursor = connection.cursor()
cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
# 執行需要最高隔離級別的操作陷阱 3：過度使用悲觀鎖
問題：過度使用 select_for_update() 會嚴重影響並發性能。
解決方案：
# 使用樂觀鎖替代
class Product(models.Model):
name = models.CharField(max_length=100)
stock = models.IntegerField()
version = models.IntegerField(default=0) # 版本號
def optimistic_update(self, **kwargs):
"""樂觀鎖更新"""
# 記錄當前版本
current_version = self.version
# 更新資料和版本號
kwargs['version'] = current_version + 1
# 只更新版本號匹配的記錄
updated = Product.objects.filter(
id=self.id,
version=current_version
).update(**kwargs)
if not updated:
raise ConcurrentModificationError("資料已被其他用戶修改")
# 更新本地物件
for key, value in kwargs.items():
setattr(self, key, value)📋 最佳實踐總結
交易要短小精悍
- 只包含必要的資料庫操作
- 耗時操作放到交易外
選擇合適的隔離級別
- 大部分情況 Read Committed 就夠了
- 金融相關才需要 Serializable
正確組合使用工具
atomic()保證原子性select_for_update()處理關鍵資源F()表達式避免競爭條件
監控和優化
- 監控慢查詢和鎖等待
- 定期檢查死鎖日誌
- 壓力測試驗證並發處理
🎯 結語
並發控制是每個 Django 開發者必須掌握的技能。記住：
- 單用
atomic()是不夠的，它只保證原子性，不能防止並發問題 - 根據業務場景選擇合適的工具組合
- 在正確性和性能之間找到平衡
掌握這些知識，你就能構建出既正確又高效的應用系統！