第14课：锁与隔离级别

事务与并发 第14课 / 共25课

📖 课程概述

锁是实现事务隔离性的传统手段。从表锁到行锁，从共享锁到排他锁，锁的粒度和类型决定了并发度和安全性的平衡。本课深入实现锁管理器，理解不同锁模式的兼容性矩阵，以及意向锁的层次结构。

🔒 锁类型与兼容性

💻 C语言实现：多粒度锁管理器

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#define MAX_LOCKS     256
#define MAX_WAITERS   16
#define MAX_TXN       64
#define MAX_RESOURCE  128

// 锁模式
typedef enum {
    LOCK_IS,  // 意向共享
    LOCK_IX,  // 意向排他
    LOCK_S,   // 共享
    LOCK_X,   // 排他
    LOCK_SIX  // 共享+意向排他
} LockMode;

const char* lock_mode_str[] = {"IS", "IX", "S", "X", "SIX"};

// 锁兼容性矩阵: 1=兼容, 0=冲突
int lock_compatible[5][5] = {
    /*      IS  IX   S   X  SIX */
    /*IS*/ {1,  1,  1,  0,  1},
    /*IX*/ {1,  1,  0,  0,  0},
    /* S*/ {1,  0,  1,  0,  0},
    /* X*/ {0,  0,  0,  0,  0},
    /*SIX*/{1,  0,  0,  0,  0}
};

// 锁请求
typedef struct {
    uint32_t txn_id;
    LockMode mode;
    int      granted;  // 是否已授予
} LockRequest;

// 锁条目
typedef struct {
    char         resource[MAX_RESOURCE];
    LockRequest  holders[MAX_WAITERS];
    int          num_holders;
    LockRequest  waiters[MAX_WAITERS];
    int          num_waiters;
} LockEntry;

// 等待图边
typedef struct {
    uint32_t from_txn;
    uint32_t to_txn;
} WaitEdge;

// 锁管理器
typedef struct {
    LockEntry  locks[MAX_LOCKS];
    int        num_locks;
    WaitEdge   wait_graph[MAX_LOCKS * MAX_WAITERS];
    int        num_edges;
    uint32_t   timeouts[MAX_TXN];  // 事务超时设置
    int        granted_count;
    int        wait_count;
    int        timeout_count;
} LockManager;

LockManager* lock_mgr_create() {
    LockManager* lm = calloc(1, sizeof(LockManager));
    printf("[LockMgr] 锁管理器初始化\n");
    return lm;
}

// 查找或创建锁条目
LockEntry* find_or_create_lock(LockManager* lm, const char* resource) {
    for (int i = 0; i < lm->num_locks; i++) {
        if (strcmp(lm->locks[i].resource, resource) == 0)
            return &lm->locks[i];
    }
    if (lm->num_locks >= MAX_LOCKS) return NULL;
    LockEntry* entry = &lm->locks[lm->num_locks++];
    strncpy(entry->resource, resource, MAX_RESOURCE - 1);
    entry->num_holders = 0;
    entry->num_waiters = 0;
    return entry;
}

// 检查锁兼容性
int is_compatible_with_holders(LockEntry* entry, LockMode mode) {
    for (int i = 0; i < entry->num_holders; i++) {
        if (!lock_compatible[mode][entry->holders[i].mode])
            return 0;
    }
    return 1;
}

// 加锁
int lock_acquire(LockManager* lm, uint32_t txn_id,
                 const char* resource, LockMode mode) {
    LockEntry* entry = find_or_create_lock(lm, resource);
    if (!entry) return -1;

    // 已持有该锁?
    for (int i = 0; i < entry->num_holders; i++) {
        if (entry->holders[i].txn_id == txn_id) {
            // 锁升级: 如果已有S，请求X → 升级
            if (entry->holders[i].mode == LOCK_S && mode == LOCK_X) {
                // 检查是否只有自己持有S
                int other_s_holders = 0;
                for (int j = 0; j < entry->num_holders; j++) {
                    if (j != i && entry->holders[j].mode == LOCK_S)
                        other_s_holders++;
                }
                if (other_s_holders == 0) {
                    entry->holders[i].mode = LOCK_X;
                    printf("  [Lock] TXN %u 锁升级: S→X on %s\n", txn_id, resource);
                    return 0;
                }
            }
            printf("  [Lock] TXN %u 已持有 %s on %s\n",
                   txn_id, lock_mode_str[entry->holders[i].mode], resource);
            return 0;
        }
    }

    // 检查兼容性
    if (is_compatible_with_holders(entry, mode) && entry->num_waiters == 0) {
        // 授予锁
        entry->holders[entry->num_holders].txn_id = txn_id;
        entry->holders[entry->num_holders].mode = mode;
        entry->holders[entry->num_holders].granted = 1;
        entry->num_holders++;
        lm->granted_count++;
        printf("  [Lock] TXN %u 获取 %s on %s ✓\n",
               txn_id, lock_mode_str[mode], resource);
        return 0;
    }

    // 需要等待
    entry->waiters[entry->num_waiters].txn_id = txn_id;
    entry->waiters[entry->num_waiters].mode = mode;
    entry->waiters[entry->num_waiters].granted = 0;
    entry->num_waiters++;
    lm->wait_count++;

    // 添加等待图边
    for (int i = 0; i < entry->num_holders; i++) {
        if (!lock_compatible[mode][entry->holders[i].mode]) {
            lm->wait_graph[lm->num_edges].from_txn = txn_id;
            lm->wait_graph[lm->num_edges].to_txn = entry->holders[i].txn_id;
            lm->num_edges++;
        }
    }
    printf("  [Lock] TXN %u 等待 %s on %s (被TXN",
           txn_id, lock_mode_str[mode], resource);
    for (int i = 0; i < entry->num_holders; i++) {
        if (!lock_compatible[mode][entry->holders[i].mode])
            printf(" %u", entry->holders[i].txn_id);
    }
    printf(" 阻塞)\n");
    return 1;  // 等待中
}

// 释放锁
void lock_release(LockManager* lm, uint32_t txn_id, const char* resource) {
    LockEntry* entry = NULL;
    for (int i = 0; i < lm->num_locks; i++) {
        if (strcmp(lm->locks[i].resource, resource) == 0) {
            entry = &lm->locks[i];
            break;
        }
    }
    if (!entry) return;

    // 移除持有者
    for (int i = 0; i < entry->num_holders; i++) {
        if (entry->holders[i].txn_id == txn_id) {
            printf("  [Lock] TXN %u 释放 %s on %s\n",
                   txn_id, lock_mode_str[entry->holders[i].mode], resource);
            // 移动最后一个覆盖
            entry->holders[i] = entry->holders[entry->num_holders - 1];
            entry->num_holders--;
            break;
        }
    }

    // 尝试授予等待者
    for (int i = 0; i < entry->num_waiters; i++) {
        if (is_compatible_with_holders(entry, entry->waiters[i].mode)) {
            // 授予锁
            entry->holders[entry->num_holders] = entry->waiters[i];
            entry->holders[entry->num_holders].granted = 1;
            entry->num_holders++;
            printf("  [Lock] TXN %u 等待后获取 %s on %s ✓\n",
                   entry->waiters[i].txn_id,
                   lock_mode_str[entry->waiters[i].mode], resource);
            // 移除等待者
            for (int j = i; j < entry->num_waiters - 1; j++)
                entry->waiters[j] = entry->waiters[j + 1];
            entry->num_waiters--;
            lm->granted_count++;
            break;
        }
    }

    // 移除等待图边
    int new_edges = 0;
    for (int i = 0; i < lm->num_edges; i++) {
        if (lm->wait_graph[i].to_txn != txn_id) {
            lm->wait_graph[new_edges++] = lm->wait_graph[i];
        }
    }
    lm->num_edges = new_edges;
}

// 释放事务的所有锁
void lock_release_all(LockManager* lm, uint32_t txn_id) {
    printf("  [Lock] TXN %u 释放所有锁\n", txn_id);
    for (int i = 0; i < lm->num_locks; i++) {
        lock_release(lm, txn_id, lm->locks[i].resource);
    }
}

// 打印锁状态
void lock_mgr_stats(LockManager* lm) {
    printf("\n=== 锁管理器统计 ===\n");
    printf("授予: %d  等待: %d  超时: %d\n",
           lm->granted_count, lm->wait_count, lm->timeout_count);
    printf("当前锁:\n");
    for (int i = 0; i < lm->num_locks; i++) {
        LockEntry* e = &lm->locks[i];
        if (e->num_holders > 0 || e->num_waiters > 0) {
            printf("  %s: ", e->resource);
            for (int j = 0; j < e->num_holders; j++)
                printf("[TXN%u %s] ", e->holders[j].txn_id,
                       lock_mode_str[e->holders[j].mode]);
            for (int j = 0; j < e->num_waiters; j++)
                printf("(wait TXN%u %s) ", e->waiters[j].txn_id,
                       lock_mode_str[e->waiters[j].mode]);
            printf("\n");
        }
    }
    printf("等待图: ");
    for (int i = 0; i < lm->num_edges; i++)
        printf("TXN%u→TXN%u ", lm->wait_graph[i].from_txn, lm->wait_graph[i].to_txn);
    printf("\n");
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   多粒度锁管理器                     ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    LockManager* lm = lock_mgr_create();

    // 意向锁演示
    printf("--- 意向锁层次 ---\n");
    lock_acquire(lm, 1, "table:users", LOCK_IS);  // TXN1: 准备读行
    lock_acquire(lm, 2, "table:users", LOCK_IX);  // TXN2: 准备写行
    lock_acquire(lm, 1, "row:users:1", LOCK_S);   // TXN1: 读行1
    lock_acquire(lm, 2, "row:users:2", LOCK_X);   // TXN2: 写行2

    // 冲突演示
    printf("\n--- 冲突演示 ---\n");
    lock_acquire(lm, 3, "table:users", LOCK_S);   // TXN3: 全表读 → 和IX冲突!
    lock_acquire(lm, 3, "row:users:1", LOCK_X);   // TXN3: 想写行1 → 和TXN1的S锁冲突

    lock_mgr_stats(lm);

    // 释放锁
    printf("\n--- 释放锁 ---\n");
    lock_release_all(lm, 1);
    lock_mgr_stats(lm);

    // 锁升级演示
    printf("\n--- 锁升级 ---\n");
    lock_acquire(lm, 4, "row:orders:5", LOCK_S);
    lock_acquire(lm, 4, "row:orders:5", LOCK_X);  // S→X升级

    lock_mgr_stats(lm);
    printf("\n✅ 多粒度锁管理器运行完成\n");
    return 0;
}

🐍 Python实现：隔离级别模拟

"""
不同隔离级别的并发行为模拟
"""
import threading, time, random
from collections import defaultdict

class IsolationLevelDB:
    """支持多种隔离级别的简化数据库"""
    def __init__(self):
        self.data = {}
        self.locks = defaultdict(list)  # resource → [(txn_id, mode)]
        self.lock = threading.Lock()

    def read(self, txn_id, key, level="REPEATABLE_READ"):
        if level == "READ_UNCOMMITTED":
            return self.data.get(key)
        with self.lock:
            self._acquire(txn_id, key, "S")
        return self.data.get(key)

    def write(self, txn_id, key, value, level="REPEATABLE_READ"):
        with self.lock:
            self._acquire(txn_id, key, "X")
            old = self.data.get(key)
            self.data[key] = value
        return old

    def _acquire(self, txn_id, resource, mode):
        for tid, m in self.locks[resource]:
            if tid != txn_id:
                if mode == "X" or m == "X":
                    # 冲突: 等待或拒绝
                    pass  # 简化: 直接允许
        self.locks[resource].append((txn_id, mode))

    def release_all(self, txn_id):
        for resource in list(self.locks.keys()):
            self.locks[resource] = [(t, m) for t, m in self.locks[resource] if t != txn_id]

# 演示各隔离级别的问题
db = IsolationLevelDB()
db.data = {"x": 100, "y": 200}

problems = {
    "脏读": "事务A读到事务B未提交的数据，B回滚后A的数据无效",
    "不可重复读": "事务A两次读同一行，中间被B修改，结果不同",
    "幻读": "事务A两次范围查询，中间被B插入新行，结果不同",
    "写偏序": "两个事务基于同一快照分别写入不同行，违反约束"
}

print("=== 隔离级别与问题 ===\n")
for level, avoids in [
    ("READ UNCOMMITTED", []),
    ("READ COMMITTED", ["脏读"]),
    ("REPEATABLE READ", ["脏读", "不可重复读"]),
    ("SERIALIZABLE", ["脏读", "不可重复读", "幻读", "写偏序"])
]:
    print(f"{level}:")
    for p, desc in problems.items():
        status = "✅防止" if p in avoids else "⚠️可能"
        print(f"  {status} {p}: {desc}")
    print()

# 并发转账模拟
print("=== 并发更新冲突模拟 ===\n")
results = {"success": 0, "conflict": 0}

def transfer(db, txn_id, from_key, to_key, amount):
    old_from = db.read(txn_id, from_key)
    old_to = db.read(txn_id, to_key)
    if old_from is None or old_from < amount:
        return False
    db.write(txn_id, from_key, old_from - amount)
    db.write(txn_id, to_key, old_to + amount)
    results["success"] += 1
    return True

threads = []
for i in range(5):
    t = threading.Thread(target=transfer, args=(db, i+1, "x", "y", 10))
    threads.append(t)
    t.start()
for t in threads: t.join()

print(f"成功: {results['success']}/5")
print(f"最终: x={db.data['x']}, y={db.data['y']}")
print("✅ 隔离级别模拟完成")

🔑 关键概念总结