第11课 - ACID与事务 - 数据库内核课程

📖 课程概述

事务是数据库区别于普通文件系统的核心特性。ACID(原子性、一致性、隔离性、持久性)是事务的四个基本保证。本课深入讲解ACID的每一个维度，理解数据库如何在故障和并发环境下维持这些保证。

本课目标：深入理解ACID的每个维度，实现事务管理器，分析不同故障场景下的恢复策略。

🛡️ ACID详解

ACID四大保证: ┌──────────────────────────────────────────────┐ │ Atomicity (原子性) │ │ 事务要么全部执行，要么全部不执行 │ │ 实现: WAL + Undo Log │ │ ┌──────────────────────────┐ │ │ │ BEGIN → [Ops] → COMMIT │ │ │ │ ↓崩溃 │ │ │ │ → Undo回滚所有操作 │ │ │ └──────────────────────────┘ │ ├──────────────────────────────────────────────┤ │ Consistency (一致性) │ │ 事务将数据库从一个一致状态转为另一个一致状态 │ │ 实现: 约束检查 + 触发器 + 应用逻辑 │ ├──────────────────────────────────────────────┤ │ Isolation (隔离性) │ │ 并发事务互不干扰 │ │ 实现: 锁 + MVCC │ │ ┌─────────┐ ┌─────────┐ │ │ │ 事务A │ │ 事务B │ │ │ │ x=x+10 │ │ x=x-5 │ │ │ │ 隔离! │ │ 隔离! │ │ │ └─────────┘ └─────────┘ │ ├──────────────────────────────────────────────┤ │ Durability (持久性) │ │ 已提交的事务结果永久保存 │ │ 实现: WAL + fsync │ │ COMMIT → WAL刷盘 → 返回成功 │ └──────────────────────────────────────────────┘

事务状态机

┌──────────┐ │ Active │ ← 事务开始 │ 活跃 │ └──┬───┬───┘ 正常完成│ │出错/ROLLBACK ▼ ▼ ┌────────┐ ┌──────────┐ │Partial │ │Aborted │ │Commit │ │已中止 │ │部分提交│ │→ Undo回滚│ └───┬────┘ └──────────┘ WAL刷盘│ ▼ ┌──────────┐ │Committed │ ← 持久化完成 │已提交 │ └──────────┘

💻 C语言实现：事务管理器

#include <stdio.h> #include <stdlib.h> #include <string.h> #include <stdint.h> #define MAX_TXN 64 #define MAX_LOG_ENTRY 256 #define MAX_KEY 64 #define MAX_VAL 255 // 事务状态 typedef enum { TXN_ACTIVE, TXN_COMMITTED, TXN_ABORTED } TxnState; // WAL日志条目 typedef struct { uint32_t lsn; uint32_t txn_id; char op_type[16]; // INSERT/UPDATE/DELETE/COMMIT/ABORT/BEGIN char key[MAX_KEY]; char old_val[MAX_VAL]; // 旧值(用于undo) char new_val[MAX_VAL]; // 新值 } LogEntry; // WAL管理器 typedef struct { LogEntry entries[MAX_LOG_ENTRY]; int count; uint32_t next_lsn; } WALManager; void wal_init(WALManager* wal) { memset(wal, 0, sizeof(WALManager)); wal->next_lsn = 1; } void wal_append(WALManager* wal, uint32_t txn_id, const char* op, const char* key, const char* old_v, const char* new_v) { LogEntry* e = &wal->entries[wal->count++]; e->lsn = wal->next_lsn++; e->txn_id = txn_id; strncpy(e->op_type, op, 15); if (key) strncpy(e->key, key, MAX_KEY - 1); if (old_v) strncpy(e->old_val, old_v, MAX_VAL - 1); if (new_v) strncpy(e->new_val, new_v, MAX_VAL - 1); printf(" [WAL] LSN=%u TXN=%u %s", e->lsn, txn_id, op); if (key) printf(" key=%s", key); printf("\n"); } // 数据存储(简化KV) typedef struct { char key[MAX_KEY]; char value[MAX_VAL]; uint32_t version; // 简单版本号 } KVEntry; #define MAX_KV 100 typedef struct { KVEntry entries[MAX_KV]; int count; } KVStore; const char* kv_get(KVStore* store, const char* key) { for (int i = 0; i < store->count; i++) { if (strcmp(store->entries[i].key, key) == 0) return store->entries[i].value; } return NULL; } void kv_put(KVStore* store, const char* key, const char* val) { for (int i = 0; i < store->count; i++) { if (strcmp(store->entries[i].key, key) == 0) { strcpy(store->entries[i].value, val); store->entries[i].version++; return; } } strcpy(store->entries[store->count].key, key); strcpy(store->entries[store->count].value, val); store->entries[store->count].version = 1; store->count++; } void kv_delete(KVStore* store, const char* key) { for (int i = 0; i < store->count; i++) { if (strcmp(store->entries[i].key, key) == 0) { // 移动最后一个覆盖 store->entries[i] = store->entries[store->count - 1]; store->count--; return; } } } // 事务描述符 typedef struct { uint32_t txn_id; TxnState state; int start_lsn; // 事务开始的LSN索引 int last_lsn; // 最后一条日志的索引 } Transaction; // 事务管理器 typedef struct { Transaction txns[MAX_TXN]; int num_txns; uint32_t next_txn_id; WALManager wal; KVStore store; } TransactionManager; TransactionManager* txn_mgr_create() { TransactionManager* tm = calloc(1, sizeof(TransactionManager)); tm->next_txn_id = 1; wal_init(&tm->wal); printf("[TxnMgr] 事务管理器初始化完成\n"); return tm; } uint32_t txn_begin(TransactionManager* tm) { uint32_t id = tm->next_txn_id++; Transaction* txn = &tm->txns[tm->num_txns++]; txn->txn_id = id; txn->state = TXN_ACTIVE; txn->start_lsn = tm->wal.count; txn->last_lsn = tm->wal.count - 1; wal_append(&tm->wal, id, "BEGIN", NULL, NULL, NULL); printf("[TxnMgr] 事务 %u BEGIN\n", id); return id; } // 查找事务 Transaction* find_txn(TransactionManager* tm, uint32_t txn_id) { for (int i = 0; i < tm->num_txns; i++) { if (tm->txns[i].txn_id == txn_id) return &tm->txns[i]; } return NULL; } void txn_insert(TransactionManager* tm, uint32_t txn_id, const char* key, const char* val) { Transaction* txn = find_txn(tm, txn_id); if (!txn || txn->state != TXN_ACTIVE) return; wal_append(&tm->wal, txn_id, "INSERT", key, NULL, val); kv_put(&tm->store, key, val); txn->last_lsn = tm->wal.count - 1; } void txn_update(TransactionManager* tm, uint32_t txn_id, const char* key, const char* new_val) { Transaction* txn = find_txn(tm, txn_id); if (!txn || txn->state != TXN_ACTIVE) return; const char* old_val = kv_get(&tm->store, key); wal_append(&tm->wal, txn_id, "UPDATE", key, old_val ? old_val : "", new_val); kv_put(&tm->store, key, new_val); txn->last_lsn = tm->wal.count - 1; } void txn_delete(TransactionManager* tm, uint32_t txn_id, const char* key) { Transaction* txn = find_txn(tm, txn_id); if (!txn || txn->state != TXN_ACTIVE) return; const char* old_val = kv_get(&tm->store, key); wal_append(&tm->wal, txn_id, "DELETE", key, old_val ? old_val : "", NULL); kv_delete(&tm->store, key); txn->last_lsn = tm->wal.count - 1; } void txn_commit(TransactionManager* tm, uint32_t txn_id) { Transaction* txn = find_txn(tm, txn_id); if (!txn || txn->state != TXN_ACTIVE) return; wal_append(&tm->wal, txn_id, "COMMIT", NULL, NULL, NULL); txn->state = TXN_COMMITTED; txn->last_lsn = tm->wal.count - 1; printf("[TxnMgr] 事务 %u COMMITTED ✓\n", txn_id); } void txn_abort(TransactionManager* tm, uint32_t txn_id) { Transaction* txn = find_txn(tm, txn_id); if (!txn || txn->state != TXN_ACTIVE) return; // Undo: 逆序回放日志 printf("[TxnMgr] 事务 %u ABORT → Undo回滚:\n", txn_id); for (int i = txn->last_lsn; i >= txn->start_lsn; i--) { LogEntry* e = &tm->wal.entries[i]; if (e->txn_id != txn_id) continue; if (strcmp(e->op_type, "INSERT") == 0) { kv_delete(&tm->store, e->key); printf(" Undo INSERT: 删除 %s\n", e->key); } else if (strcmp(e->op_type, "UPDATE") == 0) { kv_put(&tm->store, e->key, e->old_val); printf(" Undo UPDATE: 恢复 %s=%s\n", e->key, e->old_val); } else if (strcmp(e->op_type, "DELETE") == 0) { kv_put(&tm->store, e->key, e->old_val); printf(" Undo DELETE: 恢复 %s=%s\n", e->key, e->old_val); } } wal_append(&tm->wal, txn_id, "ABORT", NULL, NULL, NULL); txn->state = TXN_ABORTED; printf("[TxnMgr] 事务 %u ABORTED ✗\n", txn_id); } // 崩溃恢复(Redo) void crash_recovery(TransactionManager* tm) { printf("\n[Recovery] 开始崩溃恢复...\n"); // 分析阶段：找出所有已提交和未提交的事务 uint32_t committed[MAX_TXN], aborted[MAX_TXN]; int n_committed = 0, n_aborted = 0; for (int i = 0; i < tm->wal.count; i++) { LogEntry* e = &tm->wal.entries[i]; if (strcmp(e->op_type, "COMMIT") == 0) committed[n_committed++] = e->txn_id; else if (strcmp(e->op_type, "ABORT") == 0) aborted[n_aborted++] = e->txn_id; } // Redo阶段：重做所有已提交事务的操作 printf("[Recovery] Redo阶段:\n"); for (int i = 0; i < tm->wal.count; i++) { LogEntry* e = &tm->wal.entries[i]; int is_committed = 0; for (int j = 0; j < n_committed; j++) { if (committed[j] == e->txn_id) { is_committed = 1; break; } } if (is_committed && strcmp(e->op_type, "INSERT") == 0) { kv_put(&tm->store, e->key, e->new_val); printf(" Redo INSERT: %s=%s\n", e->key, e->new_val); } else if (is_committed && strcmp(e->op_type, "UPDATE") == 0) { kv_put(&tm->store, e->key, e->new_val); printf(" Redo UPDATE: %s=%s\n", e->key, e->new_val); } else if (is_committed && strcmp(e->op_type, "DELETE") == 0) { kv_delete(&tm->store, e->key); printf(" Redo DELETE: %s\n", e->key); } } // Undo阶段：回滚未提交事务 printf("[Recovery] Undo阶段:\n"); for (int i = tm->wal.count - 1; i >= 0; i--) { LogEntry* e = &tm->wal.entries[i]; int is_committed = 0, is_aborted = 0; for (int j = 0; j < n_committed; j++) if (committed[j] == e->txn_id) is_committed = 1; for (int j = 0; j < n_aborted; j++) if (aborted[j] == e->txn_id) is_aborted = 1; if (!is_committed && !is_aborted && (strcmp(e->op_type, "INSERT") == 0)) { kv_delete(&tm->store, e->key); printf(" Undo INSERT: 删除 %s\n", e->key); } } printf("[Recovery] 恢复完成\n"); } void print_store(KVStore* store) { printf("\n=== 当前数据 ===\n"); for (int i = 0; i < store->count; i++) { printf(" %s = %s (v%u)\n", store->entries[i].key, store->entries[i].value, store->entries[i].version); } } int main() { printf("╔══════════════════════════════════════╗\n"); printf("║ ACID事务管理器 ║\n"); printf("╚══════════════════════════════════════╝\n\n"); TransactionManager* tm = txn_mgr_create(); // 事务1: 正常提交 printf("--- 事务1: 正常流程 ---\n"); uint32_t t1 = txn_begin(tm); txn_insert(tm, t1, "alice", "Beijing"); txn_insert(tm, t1, "bob", "Shanghai"); txn_update(tm, t1, "alice", "Hangzhou"); txn_commit(tm, t1); // 事务2: 中止回滚 printf("\n--- 事务2: 中止回滚 ---\n"); uint32_t t2 = txn_begin(tm); txn_update(tm, t2, "alice", "Shenzhen"); txn_delete(tm, t2, "bob"); txn_abort(tm, t2); print_store(&tm->store); // 事务3: 模拟崩溃后恢复 printf("\n--- 事务3+4: 崩溃恢复模拟 ---\n"); uint32_t t3 = txn_begin(tm); txn_insert(tm, t3, "charlie", "Guangzhou"); txn_commit(tm, t3); uint32_t t4 = txn_begin(tm); txn_update(tm, t4, "alice", "Wuhan"); txn_insert(tm, t4, "diana", "Chengdu"); // t4未提交! 模拟崩溃 printf("\n[模拟崩溃] 事务4未提交!\n"); crash_recovery(tm); print_store(&tm->store); printf("\n✅ ACID事务管理器运行完成\n"); return 0; }

🐍 Python实现：银行转账模拟

""" 银行转账 - ACID的经典演示原子性：转出和转入必须同时成功或同时失败 """ import time, threading from dataclasses import dataclass, field from typing import Dict, Optional @dataclass class Account: id: str balance: float version: int = 0 class WALLog: def __init__(self): self.entries = [] def append(self, txn_id, op, **data): self.entries.append({"txn": txn_id, "op": op, **data, "ts": time.time()}) def get_txn_ops(self, txn_id): return [e for e in self.entries if e["txn"] == txn_id] class BankSystem: def __init__(self): self.accounts: Dict[str, Account] = {} self.wal = WALLog() self.lock = threading.Lock() self.next_txn = 1 def create_account(self, account_id: str, balance: float): self.accounts[account_id] = Account(id=account_id, balance=balance) print(f" 创建账户 {account_id}: ¥{balance:.2f}") def transfer(self, from_id: str, to_id: str, amount: float) -> bool: """原子转账""" txn_id = self.next_txn self.next_txn += 1 print(f"\n[Txn {txn_id}] 转账: {from_id} → {to_id} ¥{amount:.2f}") with self.lock: # 1. 检查账户存在 if from_id not in self.accounts or to_id not in self.accounts: print(f" [Txn {txn_id}] 失败: 账户不存在") return False from_acc = self.accounts[from_id] to_acc = self.accounts[to_id] # 2. 记录旧值到WAL old_from = from_acc.balance old_to = to_acc.balance self.wal.append(txn_id, "DEBIT", account=from_id, old=old_from, new=old_from-amount) self.wal.append(txn_id, "CREDIT", account=to_id, old=old_to, new=old_to+amount) # 3. 检查余额 if from_acc.balance < amount: print(f" [Txn {txn_id}] 失败: 余额不足 (¥{from_acc.balance:.2f})") self.wal.append(txn_id, "ABORT") return False # 4. 执行转账(原子操作) from_acc.balance -= amount from_acc.version += 1 to_acc.balance += amount to_acc.version += 1 # 5. 提交 self.wal.append(txn_id, "COMMIT") print(f" [Txn {txn_id}] 成功: {from_id}=¥{from_acc.balance:.2f}, {to_id}=¥{to_acc.balance:.2f}") return True def audit(self): total = sum(a.balance for a in self.accounts.values()) print(f"\n=== 审计 ===") for acc in self.accounts.values(): print(f" {acc.id}: ¥{acc.balance:.2f} (v{acc.version})") print(f" 总额: ¥{total:.2f}") def concurrent_transfers(self, transfers, num_threads=4): """并发转账测试""" results = [] def do_transfer(from_id, to_id, amount): r = self.transfer(from_id, to_id, amount) results.append(r) threads = [] for from_id, to_id, amount in transfers: t = threading.Thread(target=do_transfer, args=(from_id, to_id, amount)) threads.append(t) t.start() if len(threads) >= num_threads: for t in threads: t.join() threads = [] for t in threads: t.join() return results # 测试 bank = BankSystem() bank.create_account("Alice", 1000) bank.create_account("Bob", 2000) bank.create_account("Charlie", 500) # 正常转账 bank.transfer("Alice", "Bob", 200) bank.transfer("Bob", "Charlie", 500) # 余额不足 bank.transfer("Charlie", "Alice", 1000) # 审计 bank.audit() # 并发测试 print("\n=== 并发转账测试 ===") transfers = [("Alice","Bob",100)]*5 + [("Bob","Charlie",200)]*5 results = bank.concurrent_transfers(transfers) print(f"成功: {sum(results)}/{len(results)}") bank.audit() print("\n✅ ACID银行转账模拟完成")

📊 ACID实现代价分析

ACID保证的实现代价: 保证实现机制性能影响优化手段 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 原子性 Undo Log 2x写开销组提交一致性约束检查 CPU开销延迟检查隔离性锁/MVCC 等待开销细粒度锁持久性 WAL+fsync I/O延迟异步刷盘关键权衡: - 隔离级别越高 → 性能越差 - fsync频率越高 → 持久性越强 → 延迟越大 - 事务粒度越细 → 并发越高 → 逻辑越复杂

🔑 关键概念总结

原子性：事务不可分割，Undo Log保证可回滚
一致性：事务保持数据完整性约束
隔离性：并发事务互不干扰，锁和MVCC实现
持久性：提交结果永久保存，WAL+fsync保证
恢复策略：Redo重做已提交事务，Undo回滚未提交事务

📝 练习

实现嵌套事务(Savepoint)，支持部分回滚
实现组提交(Group Commit)，将多个事务的WAL合并一次fsync
模拟各种故障场景(写WAL后崩溃、写数据后崩溃)，验证恢复正确性
测量不同fsync策略(always/batch/none)对TPS的影响

🛡️

🏆 成就解锁：事务守护者

掌握ACID原理，你已理解数据库事务的核心保证！

✅ 原子性实现 · ✅ WAL日志 · ✅ 崩溃恢复

第11课：ACID与事务