数据库（三）：事务与并发控制——ACID、隔离级别与锁机制

1
2
3
4

BEGIN;
    UPDATE accounts SET balance = balance - 500 WHERE account_id = 1;
    UPDATE accounts SET balance = balance + 500 WHERE account_id = 2;
COMMIT;

1
2
3
4
5

-- 若无原子性：服务器在两条语句之间崩溃
UPDATE inventory SET stock = stock - 1 WHERE product_id = 42;
-- 此处发生崩溃
INSERT INTO order_items (order_id, product_id, quantity) VALUES (101, 42, 1);
-- 库存已扣减，但订单明细未创建 → 库存泄漏（inventory leak）

1
2
3
4

-- 若无一致性检查：
INSERT INTO orders (order_id, user_id) VALUES (999, 12345);
-- user_id 12345 在 users 表中并不存在
-- 现在我们拥有一条“孤儿”订单（orphaned order），无关联用户

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- 事务 A：读取账户余额
SELECT balance FROM accounts WHERE account_id = 1;  -- 返回 1000

-- 事务 B：取款 500 并提交
UPDATE accounts SET balance = balance - 500 WHERE account_id = 1;
COMMIT;

-- 事务 A：再次读取（同一事务内）
SELECT balance FROM accounts WHERE account_id = 1;  -- 返回 500！
-- 余额在事务中途被修改 → 事务 A 对世界的视图已不一致。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

-- 启动事务
BEGIN;
-- 或：START TRANSACTION;

-- 执行操作
INSERT INTO orders (user_id, status) VALUES (1, 'pending');
UPDATE inventory SET stock = stock - 1 WHERE product_id = 42;

-- 创建保存点（savepoint）——支持部分回滚
SAVEPOINT after_inventory;

-- 继续操作
INSERT INTO shipping (order_id, address) VALUES (currval('orders_order_id_seq'), '123 Main St');

-- 哎呀，地址错了。回滚至保存点。
ROLLBACK TO SAVEPOINT after_inventory;

-- 修正后重试
INSERT INTO shipping (order_id, address) VALUES (currval('orders_order_id_seq'), '456 Oak Ave');

-- 提交全部变更
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 会话 1                          -- 会话 2
BEGIN;
UPDATE products SET price = 0.01
WHERE product_id = 1;
                                      BEGIN;
                                      -- 脏读：看到 price = 0.01
                                      SELECT price FROM products
                                      WHERE product_id = 1;
ROLLBACK;  -- price 恢复为 49.99
                                      -- 会话 2 基于 price 0.01 做出了决策，
                                      -- 但该价格从未真实存在过。
                                      COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 会话 1                          -- 会话 2
BEGIN;
SELECT balance FROM accounts
WHERE id = 1;  -- 返回 1000
                                      BEGIN;
                                      UPDATE accounts SET balance = 500
                                      WHERE id = 1;
                                      COMMIT;
SELECT balance FROM accounts
WHERE id = 1;  -- 返回 500！
-- 同一查询，在同一事务内返回不同结果。
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 会话 1                          -- 会话 2
BEGIN;
SELECT COUNT(*) FROM orders
WHERE status = 'pending';  -- 返回 5
                                      BEGIN;
                                      INSERT INTO orders (user_id, status)
                                      VALUES (99, 'pending');
                                      COMMIT;
SELECT COUNT(*) FROM orders
WHERE status = 'pending';  -- 返回 6！
-- 新增了一行（幻影行）。
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- 为单个事务设置隔离级别
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
-- ... your queries ...
COMMIT;

-- 为会话设置默认隔离级别（PostgreSQL）
SET default_transaction_isolation = 'read committed';

-- 查看当前会话默认隔离级别（PostgreSQL）
SHOW default_transaction_isolation;

-- MySQL
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT @@transaction_isolation;

1
2
3
4
5
6
7
8

-- 事务 100 插入一行
INSERT INTO accounts (id, balance) VALUES (1, 1000);
-- 行：xmin=100, xmax=0, balance=1000

-- 事务 200 更新该行
UPDATE accounts SET balance = 500 WHERE id = 1;
-- 旧行：xmin=100, xmax=200, balance=1000  （对旧快照仍可见）
-- 新行：xmin=200, xmax=0,   balance=500   （对新快照可见）

1
2
3
4
5
6
7

-- SELECT FOR UPDATE 获取行锁
-- 其他事务若尝试更新同一行，将被阻塞等待
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
-- 此行已被锁定。其他事务若尝试修改它，将等待。
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;  -- 锁释放

1
2
3
4
5
6

-- SELECT FOR SHARE（共享读锁）
-- 其他事务可读，但不可修改
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR SHARE;
-- 此行被读锁保护。其他事务可读，但不能 UPDATE/DELETE。
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- SKIP LOCKED：非阻塞队列模式
-- 适用于任务队列——工作者（worker）获取未被锁定的任务
BEGIN;
SELECT * FROM tasks
WHERE status = 'pending'
ORDER BY created_at
LIMIT 1
FOR UPDATE SKIP LOCKED;
-- 若某行已被其他 worker 锁定，则跳过，获取下一条

UPDATE tasks SET status = 'processing', worker_id = 'worker-3'
WHERE task_id = ...;  -- 上述 SELECT 返回的 ID
COMMIT;

1
2
3
4

-- NOWAIT：立即失败，而非等待
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE NOWAIT;
-- 若已被锁定，立即报错：ERROR: could not obtain lock on row

1
2
3
4
5

-- 显式表锁（PostgreSQL）
LOCK TABLE accounts IN EXCLUSIVE MODE;

-- ACCESS EXCLUSIVE：阻塞一切操作，DROP TABLE、ALTER TABLE 所需
-- ACCESS SHARE：与除 ACCESS EXCLUSIVE 外的所有锁兼容

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- 获取顾问锁（PostgreSQL）
-- 锁编号任意，由你的应用定义其含义
SELECT pg_advisory_lock(12345);
-- ... 执行需要独占访问的工作 ...
SELECT pg_advisory_unlock(12345);

-- 尝试获取（不阻塞）
SELECT pg_try_advisory_lock(12345);  -- 返回 true/false

-- 会话级顾问锁（会话结束时自动释放）
SELECT pg_advisory_lock(hashtext('process_daily_report'));

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- 事务 A                      -- 事务 B
BEGIN;                                BEGIN;
UPDATE accounts SET balance = 900
WHERE id = 1;  -- 锁定第 1 行
                                      UPDATE accounts SET balance = 1100
                                      WHERE id = 2;  -- 锁定第 2 行

UPDATE accounts SET balance = 1100
WHERE id = 2;  -- 等待事务 B 的锁
                                      UPDATE accounts SET balance = 900
                                      WHERE id = 1;  -- 等待事务 A 的锁

-- 死锁！双方都在等待对方释放锁。

1
2
3
4
5
6

ERROR:  deadlock detected
DETAIL: Process 12345 waits for ShareLock on transaction 67890;
        blocked by process 67891.
        Process 67891 waits for ShareLock on transaction 12345;
        blocked by process 12345.
HINT:   See server log for query details.

1
2
3
4
5
6
7

-- 始终先锁定较小的 account_id
BEGIN;
SELECT * FROM accounts WHERE id = LEAST(1, 2) FOR UPDATE;
SELECT * FROM accounts WHERE id = GREATEST(1, 2) FOR UPDATE;
UPDATE accounts SET balance = balance - 500 WHERE id = 1;
UPDATE accounts SET balance = balance + 500 WHERE id = 2;
COMMIT;

1
2
3
4
5

-- PostgreSQL：在锁竞争时快速失败
SET lock_timeout = '5s';
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
-- 若被锁超过 5 秒，则中止，而非无限等待

1
2
3
4
5
6

-- 悲观锁：先锁定行
BEGIN;
SELECT * FROM products WHERE product_id = 42 FOR UPDATE;
-- 检查库存、计算等
UPDATE products SET stock = stock - 1 WHERE product_id = 42;
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- 添加版本列
ALTER TABLE products ADD COLUMN version INT NOT NULL DEFAULT 1;

-- 读取当前状态（无锁）
SELECT product_id, stock, version FROM products WHERE product_id = 42;
-- 返回：stock = 10, version = 5

-- 应用层进行计算...

-- 带版本检查的更新
UPDATE products
SET stock = stock - 1, version = version + 1
WHERE product_id = 42 AND version = 5;
-- 若受影响行数为 0，说明他人已修改该行 → 需重试。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 应用层乐观锁模式
def purchase_product(product_id: int, quantity: int):
    max_retries = 3
    for attempt in range(max_retries):
        # 读取当前状态
        product = db.query(
            "SELECT stock, version FROM products WHERE product_id = %s",
            [product_id]
        )

        if product.stock < quantity:
            raise InsufficientStockError()

        # 尝试带版本检查的更新
        rows_affected = db.execute(
            """UPDATE products
               SET stock = stock - %s, version = version + 1
               WHERE product_id = %s AND version = %s""",
            [quantity, product_id, product.version]
        )

        if rows_affected == 1:
            return  # 成功
        # 版本不匹配 → 重试
    raise ConflictError("Too many retries")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

BEGIN;

INSERT INTO orders (user_id, total) VALUES (1, 99.99);

SAVEPOINT before_items;

INSERT INTO order_items (order_id, product_id, qty) VALUES (1001, 42, 1);
INSERT INTO order_items (order_id, product_id, qty) VALUES (1001, 99, 1);
-- 产品 99 缺货

ROLLBACK TO SAVEPOINT before_items;
-- 订单仍存在，商品行已撤销

INSERT INTO order_items (order_id, product_id, qty) VALUES (1001, 42, 1);

COMMIT;  -- 含 1 件商品的订单成功提交

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

async def create_order_with_fallback(conn, user_id: int, items: list[dict]):
    async with conn.transaction():
        order_id = await conn.fetchval(
            "INSERT INTO orders (user_id) VALUES ($1) RETURNING order_id", user_id
        )
        for item in items:
            try:
                async with conn.transaction():  # 自动创建保存点
                    await conn.execute(
                        "INSERT INTO order_items (order_id, product_id, qty) "
                        "VALUES ($1, $2, $3)",
                        order_id, item["product_id"], item["qty"]
                    )
            except Exception:
                pass  # 这个商品失败，外层事务继续

1
2
3
4
5
6
7
8

-- 获取排他咨询锁（如已被持有则阻塞）
SELECT pg_advisory_lock(12345);

-- 尝试获取（不阻塞，返回 true/false）
SELECT pg_try_advisory_lock(12345);

-- 释放
SELECT pg_advisory_unlock(12345);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 防止定时任务重复执行
async def run_scheduled_task(conn, task_id: int):
    acquired = await conn.fetchval("SELECT pg_try_advisory_lock($1)", task_id)
    if not acquired:
        return  # 另一个实例正在运行

    try:
        await do_expensive_work()
    finally:
        await conn.execute("SELECT pg_advisory_unlock($1)", task_id)

# 防止 "获取或创建" 模式的竞态条件
async def get_or_create_user(conn, email: str):
    lock_key = hash(email) % (2**31)
    await conn.execute("SELECT pg_advisory_lock($1)", lock_key)
    try:
        user = await conn.fetchrow("SELECT * FROM users WHERE email = $1", email)
        if user is None:
            user = await conn.fetchrow(
                "INSERT INTO users (email) VALUES ($1) RETURNING *", email
            )
        return user
    finally:
        await conn.execute("SELECT pg_advisory_unlock($1)", lock_key)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- 医生值班调度：至少一名医生必须在岗
-- 两名医生同时尝试下班

-- 事务 A（Smith 医生）：
BEGIN ISOLATION LEVEL SERIALIZABLE;
SELECT count(*) FROM doctors WHERE on_call = true;  -- 看到 2
UPDATE doctors SET on_call = false WHERE name = 'Smith';
COMMIT;  -- 成功

-- 事务 B（Jones 医生），并发执行：
BEGIN ISOLATION LEVEL SERIALIZABLE;
SELECT count(*) FROM doctors WHERE on_call = true;  -- 看到 2
UPDATE doctors SET on_call = false WHERE name = 'Jones';
COMMIT;  -- 串行化失败！回滚。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

-- 安全转账函数（PostgreSQL）
CREATE OR REPLACE FUNCTION transfer(
    from_account INT,
    to_account INT,
    amount DECIMAL(12, 2)
) RETURNS VOID AS $$
DECLARE
    from_balance DECIMAL(12, 2);
BEGIN
    -- 按一致顺序锁定两个账户（ID 小者优先）
    -- 此举可预防死锁
    IF from_account < to_account THEN
        PERFORM 1 FROM accounts WHERE account_id = from_account FOR UPDATE;
        PERFORM 1 FROM accounts WHERE account_id = to_account FOR UPDATE;
    ELSE
        PERFORM 1 FROM accounts WHERE account_id = to_account FOR UPDATE;
        PERFORM 1 FROM accounts WHERE account_id = from_account FOR UPDATE;
    END IF;

    -- 锁定后检查余额（防止竞态条件）
    SELECT balance INTO from_balance
    FROM accounts WHERE account_id = from_account;

    IF from_balance < amount THEN
        RAISE EXCEPTION 'Insufficient funds: balance=%, amount=%',
            from_balance, amount;
    END IF;

    -- 执行转账
    UPDATE accounts SET balance = balance - amount
    WHERE account_id = from_account;

    UPDATE accounts SET balance = balance + amount
    WHERE account_id = to_account;

    -- 记录转账日志
    INSERT INTO transfer_log (from_account, to_account, amount, transferred_at)
    VALUES (from_account, to_account, amount, NOW());
END;
$$ LANGUAGE plpgsql;

1
2
3
4

-- 调用方式
BEGIN;
SELECT transfer(1, 2, 500.00);
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

-- PostgreSQL：查看当前锁状态
SELECT
    l.pid,
    a.usename,
    l.locktype,
    l.relation::regclass AS table_name,
    l.mode,
    l.granted,
    a.query,
    age(now(), a.query_start) AS query_age
FROM pg_locks l
JOIN pg_stat_activity a ON l.pid = a.pid
WHERE NOT l.granted
ORDER BY a.query_start;

-- 查找阻塞查询
SELECT
    blocked.pid AS blocked_pid,
    blocked.query AS blocked_query,
    blocking.pid AS blocking_pid,
    blocking.query AS blocking_query,
    age(now(), blocked.query_start) AS waiting_time
FROM pg_stat_activity blocked
JOIN pg_locks blocked_locks ON blocked.pid = blocked_locks.pid
JOIN pg_locks blocking_locks ON blocked_locks.locktype = blocking_locks.locktype
    AND blocked_locks.relation = blocking_locks.relation
    AND blocked_locks.pid != blocking_locks.pid
JOIN pg_stat_activity blocking ON blocking_locks.pid = blocking.pid
WHERE NOT blocked_locks.granted;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- MySQL：查看 InnoDB 锁等待
SELECT
    r.trx_id AS waiting_trx,
    r.trx_mysql_thread_id AS waiting_thread,
    r.trx_query AS waiting_query,
    b.trx_id AS blocking_trx,
    b.trx_mysql_thread_id AS blocking_thread,
    b.trx_query AS blocking_query
FROM information_schema.innodb_lock_waits w
JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_id
JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id;