乐观锁：贴个标签（版本号），走的时候检查
悲观锁：直接锁门，别人进不来

-- MySQL 5.x 语法
SELECT ... FOR UPDATE;                    -- 加排他锁（X 锁）

-- MySQL 8.0 新增语法
SELECT ... FOR SHARE;                     -- 加共享锁（等同于 LOCK IN SHARE MODE）
SELECT ... FOR UPDATE NOWAIT;            -- 不等待，立即返回错误
SELECT ... FOR UPDATE SKIP LOCKED;       -- 跳过被锁的行，只处理未锁定的行
SELECT ... FOR UPDATE WAIT 5;            -- 等待 5 秒（需在 MySQL 8.0 + 某些存储引擎中）

-- 给 id=1 的账户加排他锁
SELECT * FROM account WHERE id = 1 FOR UPDATE;

-- 这时其他事务：
-- ❌ 不能 SELECT FOR UPDATE id=1
-- ❌ 不能 UPDATE/DELETE id=1
-- ✅ 可以普通 SELECT（快照读，不加锁）

-- 电商扣库存（防止超卖）
BEGIN;

-- 1. 查到库存并锁定该行
SELECT stock FROM inventory WHERE product_id = 100 FOR UPDATE;
-- → stock = 5

-- 2. 检查库存
-- 5 > 0，可以扣减

-- 3. 扣减库存
UPDATE inventory SET stock = stock - 1 WHERE product_id = 100;

COMMIT;
-- 事务提交后才释放锁，其他事务才能继续操作

-- 转账操作
BEGIN;

-- 1. 同时锁定转出和转入账户
SELECT balance FROM account WHERE id = 1 FOR UPDATE;  -- 锁转出账户
SELECT balance FROM account WHERE id = 2 FOR UPDATE;  -- 锁转入账户

-- 2. 检查余额
-- 余额足够

-- 3. 更新
UPDATE account SET balance = balance - 100 WHERE id = 1;
UPDATE account SET balance = balance + 100 WHERE id = 2;

COMMIT;

-- 在 RR 下，FOR UPDATE 使用 Next-Key Lock
SELECT * FROM order WHERE amount > 100 FOR UPDATE;
-- 锁住 amount > 100 的所有记录及其间隙
-- 其他事务无法在间隙中插入新记录

-- 在 RC 下，FOR UPDATE 使用 Record Lock
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT * FROM order WHERE amount > 100 FOR UPDATE;
-- 只锁住满足条件的记录
-- 间隙不锁，其他事务可以插入新记录

-- 如果锁被占用，立即返回错误，不等待
SELECT * FROM account WHERE id = 1 FOR UPDATE NOWAIT;
-- 行未被锁定 → 正常返回
-- 行被锁定 → 立即报错，不阻塞

-- 跳过被锁定的行，只返回未锁定的行
SELECT * FROM order WHERE status = 'PENDING' FOR UPDATE SKIP LOCKED;
-- 假设 id=1 的行被其他事务锁定
-- 本查询返回 id=2,3,5... 等未锁定的行

-- ❌ 错误：不开启事务，FOR UPDATE 的锁立即释放
SELECT * FROM account WHERE id = 1 FOR UPDATE;
-- 锁只在语句执行期间存在

-- ✅ 正确：开启事务
BEGIN;
SELECT * FROM account WHERE id = 1 FOR UPDATE;
-- 事务提交前锁一直持有
COMMIT;

-- ❌ name 列没有索引
SELECT * FROM user WHERE name = 'Alice' FOR UPDATE;
-- InnoDB 会锁全表（聚簇索引所有行）

-- ✅ 加上索引
CREATE INDEX idx_name ON user(name);
SELECT * FROM user WHERE name = 'Alice' FOR UPDATE;
-- 只锁 Alice 对应的行

-- 两个事务互相等待 → 死锁
-- 解决方法：固定加锁顺序（见上一篇文章）

-- 设置锁等待超时时间（默认 50 秒）
SET innodb_lock_wait_timeout = 5;

悲观锁：先加锁再操作 → "防止别人改"
乐观锁：直接操作，提交时检查 → "你改我就放弃/重试"

CREATE TABLE account (
    id INT PRIMARY KEY,
    balance DECIMAL(10,2),
    version INT DEFAULT 0  -- 版本号
);

-- 1. 读取数据，获取 version
SELECT id, balance, version FROM account WHERE id = 1;
-- → id=1, balance=100, version=5

-- 2. 业务计算
-- 新余额 = 100 - 20 = 80

-- 3. 更新时检查 version
UPDATE account 
SET balance = 80, version = version + 1
WHERE id = 1 AND version = 5;  -- ⚠️ 关键：version 必须匹配
-- 
-- ┌─ 如果成功（影响行数 = 1）→ 更新成功
-- └─ 如果失败（影响行数 = 0）→ 数据已被其他事务修改，重试或放弃

# Python 乐观锁示例
def update_balance(account_id, amount, max_retries=3):
    for retry in range(max_retries):
        # 1. 读取数据
        cursor.execute(
            "SELECT id, balance, version FROM account WHERE id = %s",
            (account_id,)
        )
        row = cursor.fetchone()
        current_balance = row['balance']
        current_version = row['version']

        new_balance = current_balance - amount

        # 2. 尝试更新（带 version 条件）
        cursor.execute(
            "UPDATE account SET balance = %s, version = version + 1 "
            "WHERE id = %s AND version = %s",
            (new_balance, account_id, current_version)
        )

        # 3. 检查是否更新成功
        if cursor.rowcount > 0:
            connection.commit()
            return True  # 更新成功
        else:
            # 版本冲突，重试
            connection.rollback()
            if retry == max_retries - 1:
                raise Exception("更新失败，重试次数用完")

CREATE TABLE article (
    id INT PRIMARY KEY,
    title VARCHAR(200),
    content TEXT,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- 1. 读取
SELECT id, title, content, updated_at FROM article WHERE id = 1;
-- → title='MySQL入门', updated_at='2024-01-01 10:00:00'

-- 2. 编辑、更新
UPDATE article 
SET title = 'MySQL进阶', content = '新内容', updated_at = NOW()
WHERE id = 1 AND updated_at = '2024-01-01 10:00:00';
-- 如果其他事务先更新了，updated_at 不一样，本更新会失败

-- 库存扣减（不超卖）
UPDATE inventory 
SET stock = stock - 1 
WHERE id = 1 AND stock >= 1;
-- 如果 stock = 0，影响行数为 0，说明扣减失败

-- 余额扣减
UPDATE account 
SET balance = balance - 100 
WHERE id = 1 AND balance >= 100;
-- 余额不足时更新失败

-- ❌ 事务 A：先改 id=1，再改 id=2
BEGIN;
UPDATE account SET balance = balance - 100 WHERE id = 1;
UPDATE account SET balance = balance + 100 WHERE id = 2;
COMMIT;

-- ❌ 事务 B：先改 id=2，再改 id=1
BEGIN;
UPDATE account SET balance = balance - 50 WHERE id = 2;
UPDATE account SET balance = balance + 50 WHERE id = 1;
COMMIT;
-- ⚠️ 锁获取顺序相反，容易死锁

-- ✅ 约定：所有事务按 ID 升序获取锁
-- 事务 A
BEGIN;
UPDATE account SET balance = balance - 100 WHERE id = 1;
UPDATE account SET balance = balance + 100 WHERE id = 2;
COMMIT;

-- 事务 B（也按升序）
BEGIN;
UPDATE account SET balance = balance + 50 WHERE id = 1;
UPDATE account SET balance = balance - 50 WHERE id = 2;
COMMIT;
-- ✅ 总是先锁 id=1，再锁 id=2，不会形成环路

// Java 示例：按升序锁定账户
public void transfer(Long fromId, Long toId, BigDecimal amount) {
    // 按 ID 排序，确保加锁顺序一致
    Long first = Math.min(fromId, toId);
    Long second = Math.max(fromId, toId);

    accountDao.update(first, -amount);   // 先锁小的
    accountDao.update(second, +amount);  // 再锁大的
}

-- ❌ 错误：事务中混入非数据库操作
BEGIN;
SELECT * FROM account WHERE id = 1 FOR UPDATE;
// ... 调用 HTTP API 获取外部信息（耗时很长！）
UPDATE account SET balance = balance - 100 WHERE id = 1;
COMMIT;

-- ✅ 正确：先准备好所有数据再开启事务
// 先调用外部 API（事务外）
externalData = httpClient.getExternalData();

BEGIN;
SELECT * FROM account WHERE id = 1 FOR UPDATE;
UPDATE account SET balance = balance - 100 WHERE id = 1;
COMMIT;

-- REPEATABLE READ（默认）：当前读使用 Next-Key Lock，锁范围大
-- READ COMMITTED：当前读使用 Record Lock，锁范围小

-- 如果业务允许，使用 READ COMMITTED
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;

-- ❌ 错误：一个大事务包含了所有操作
BEGIN;
INSERT INTO order ...;
INSERT INTO order_item ...;   -- 多行
UPDATE inventory ...;          -- 多个 SKU
UPDATE user_coupon ...;
UPDATE account ...;
COMMIT;

-- ✅ 正确：分解为多个小事务
BEGIN;
INSERT INTO order ...;          -- 小事务
INSERT INTO order_item ...;
COMMIT;

BEGIN;
UPDATE inventory SET stock = stock - 1 WHERE id = 123;  -- 小事务
COMMIT;

BEGIN;
UPDATE account SET balance = balance - 100 WHERE id = 1;  -- 小事务
COMMIT;

-- ❌ 没有索引：会扫描全表，所有行都加锁
SELECT * FROM user WHERE name = 'Alice' FOR UPDATE;

-- ✅ 有索引：只锁住匹配的行
CREATE INDEX idx_name ON user(name);
SELECT * FROM user WHERE name = 'Alice' FOR UPDATE;

-- ❌ 范围查询：使用间隙锁，锁定的范围大
SELECT * FROM order WHERE amount BETWEEN 100 AND 200 FOR UPDATE;

-- ✅ 等值查询+单独处理：逐条锁定，范围小
SELECT * FROM order WHERE amount = 100 FOR UPDATE;

-- 锁等待超时（默认 50 秒）
SET innodb_lock_wait_timeout = 5;  -- 降低到 5 秒

-- 应用层重试（见上一篇文章的代码示例）

事务 A：                     事务 B：
BEGIN;                       BEGIN;

UPDATE account SET balance   UPDATE account SET balance
= balance - 100              = balance - 50
WHERE id = 1;                WHERE id = 2;
-- 获得 id=1 的 Record Lock  -- 获得 id=2 的 Record Lock

UPDATE account SET balance   UPDATE account SET balance
= balance + 100              = balance + 50
WHERE id = 2;                WHERE id = 1;
-- 等待事务 B 释放 id=2      -- 等待事务 A 释放 id=1

         ⚠️ 互相等待 → 死锁！

等待图结构：
- 节点：每个事务
- 边：事务 T1 等待事务 T2 释放的锁资源

检测算法：
- 在等待图中查找"环"
- 如果存在环 → 发生死锁

-- 当 InnoDB 检测到事务等待锁时
-- 它会构建当前的等待图：
-- T1 → 等待锁资源 R2 → T2
-- T2 → 等待锁资源 R1 → T1
-- 发现环：T1 → T2 → T1
-- → 判断存在死锁

-- 查看死锁检测是否开启（默认 ON）
SHOW VARIABLES LIKE 'innodb_deadlock_detect';
-- → ON

-- 关闭死锁检测（高并发场景可能用于提升性能，但风险极大）
SET GLOBAL innodb_deadlock_detect = OFF;

-- 选择规则：
-- 1. 选择"回滚代价最小"的事务
-- 2. 代价计算基于事务修改的行数、持有的锁数量等

-- 被选中的事务会收到错误：
ERROR 1213 (40001): Deadlock found when trying to get lock;
try restarting transaction

-- 事务 A（被选为牺牲品）
BEGIN;
UPDATE account SET balance = balance - 100 WHERE id = 1;  -- 获得锁
-- 事务 B 开始执行
UPDATE account SET balance = balance + 100 WHERE id = 2;  -- 等待事务 B 的锁
-- InnoDB 检测到死锁，选择回滚事务 A

-- 事务 A 收到：
ERROR 1213 (40001): Deadlock found when trying to get lock;
try restarting transaction

-- 事务 A 自动回滚，释放 id=1 的锁
-- 事务 B 获得 id=1 的锁，继续执行

// Java 示例：死锁重试
int retryCount = 0;
int maxRetries = 3;

while (retryCount < maxRetries) {
    try {
        // 执行事务操作
        transferMoney(fromAccount, toAccount, amount);
        break;  // 成功，跳出循环
    } catch (DeadlockLoserDataAccessException e) {
        retryCount++;
        if (retryCount == maxRetries) {
            throw e;  // 重试次数耗尽，抛出异常
        }
        Thread.sleep(100);  // 等待一段时间后重试
    }
}

# Python 示例
import time
from pymysql.err import OperationalError

for retry in range(3):
    try:
        with connection.cursor() as cursor:
            cursor.execute("BEGIN")
            cursor.execute("UPDATE account SET balance = balance - 100 WHERE id = 1")
            cursor.execute("UPDATE account SET balance = balance + 100 WHERE id = 2")
            connection.commit()
        break
    except OperationalError as e:
        if e.args[0] == 1213:  # Deadlock error code
            if retry == 2:
                raise
            time.sleep(0.1 * (retry + 1))
            connection.rollback()
        else:
            raise

SHOW ENGINE INNODB STATUS\G
-- 在输出中查找 LATEST DETECTED DEADLOCK 部分

-- Performance Schema 中的死锁记录
SELECT * FROM performance_schema.events_transactions_current;

MySQL 重启后，自增 ID 的恢复方式：
SELECT MAX(auto_inc_col) FROM t FOR UPDATE;

INSERT INTO users(name) VALUES('Tom');
-- 会话层级：获取 AUTO-INC 表锁 → 分配 ID → 释放 AUTO-INC 表锁
-- 对于所有 INSERT 都使用表级锁

-- 简答的 INSERT（行数可预知）
INSERT INTO users(name) VALUES('Tom');
INSERT INTO users(name) VALUES('Jerry'), ('Bob');
-- 使用轻量互斥量（mutex），性能很好

-- 批量插入（行数不可预知）
INSERT INTO users(name) SELECT name FROM old_users;
-- 使用表级自增锁，保证连续 ID

INSERT INTO users(name) VALUES('Tom');
INSERT INTO users(name) SELECT name FROM old_users;
-- 全部使用轻量互斥量
-- 不保证 ID 连续性

BEGIN;
INSERT INTO users(name) VALUES('Tom');
  -- 1. 获取自增锁（或互斥量）
  -- 2. 分配 ID=100
  -- 3. 释放自增锁 ← 此时锁已经释放了
  -- 4. 插入数据行（持有行锁）
COMMIT;

-- 原因 1：事务回滚
BEGIN;
INSERT INTO users(name) VALUES('Tom');  -- 分配 ID=100
ROLLBACK;  -- 行被回滚，但 ID=100 已被分配
INSERT INTO users(name) VALUES('Jerry');  -- ID=101，而不是 100

-- 原因 2：批量插入预留过多
INSERT INTO users(name) VALUES('A'),('B'),('C');  -- 预留 3 个 ID
-- 实际只用了 3 个，但如果预留发生在并发环境下，可能有些 ID 被浪费

-- 原因 3：MySQL 重启
-- MySQL 重启后，自增起始值可能改变

事务 A：INSERT INTO users(name) SELECT name FROM big_table;  -- 预留 100 个 ID
事务 B：INSERT INTO users(name) VALUES('New');  -- 得到 ID=101，而不是在 A 预留的 ID 之间

-- 主库
INSERT INTO t VALUES(NULL, 'a');  -- ID=1
INSERT INTO t VALUES(NULL, 'b');  -- ID=2
-- 由于交错，可能 ID 顺序不同

-- 查看当前模式
SELECT @@innodb_autoinc_lock_mode;
-- MySQL 8.0 默认：1（consecutive）

-- 设置模式（修改配置文件 my.cnf）
-- [mysqld]
-- innodb_autoinc_lock_mode = 2

-- 查看当前自增值
SELECT AUTO_INCREMENT FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'dbname' AND TABLE_NAME = 'users';

-- 修改自增值
ALTER TABLE users AUTO_INCREMENT = 1000;

-- 事务 A 持有 id=1 的 X 锁
UPDATE users SET name = 'A' WHERE id = 1;

-- 事务 B 不能同时持有 id=1 的 X 锁
UPDATE users SET name = 'B' WHERE id = 1;  -- 等待 A

-- 事务 A：持有 id=1 的锁，等待 id=2 的锁
UPDATE users SET balance = 100 WHERE id = 1;
UPDATE users SET balance = 200 WHERE id = 2;  -- 等待 B 释放

-- 事务 B：持有 id=2 的锁，等待 id=1 的锁
UPDATE users SET balance = 300 WHERE id = 2;
UPDATE users SET balance = 400 WHERE id = 1;  -- 等待 A 释放

-- 事务 A 持有 id=1 的锁
-- 无论事务 B 愿意付出什么代价，都不能让 A 交出锁
-- A 只能在 COMMIT 或 ROLLBACK 后释放

事务 A → 等待 → 事务 B → 等待 → 事务 C → 等待 → 事务 A
             ↑                               |
             └───────────────────────────────┘

-- 经典的两事务循环等待
-- 事务 A 持有 id=1，等待 id=2
-- 事务 B 持有 id=2，等待 id=1

-- 图表示：
-- A → (等待 id=2) → B
-- B → (等待 id=1) → A
-- A ↔ B 形成循环

三个事务的循环等待：
事务 A：持有 id=1，等待 id=2
事务 B：持有 id=2，等待 id=3  
事务 C：持有 id=3，等待 id=1

等待环：A → B → C → A

-- 初始数据
-- users 表中有 id=1 和 id=2 两条记录

-- 时间点 1：事务 A 执行
BEGIN;
UPDATE users SET balance = balance - 100 WHERE id = 1;  -- 锁 id=1

-- 时间点 2：事务 B 执行
BEGIN;
UPDATE users SET balance = balance - 200 WHERE id = 2;  -- 锁 id=2

-- 时间点 3：事务 A 执行
UPDATE users SET balance = balance + 100 WHERE id = 2;
-- 等待事务 B 释放 id=2 ← Hold and Wait

-- 时间点 4：事务 B 执行
UPDATE users SET balance = balance + 200 WHERE id = 1;
-- 等待事务 A 释放 id=1 ← 循环等待形成

-- 此时：死锁！InnoDB 会检测到并处理

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id = 5 FOR UPDATE;
-- id=5 不存在，锁间隙（假设已有 id=3, id=7）

-- 事务 B
BEGIN;
SELECT * FROM users WHERE id = 6 FOR UPDATE;
-- id=6 不存在，也锁在这个间隙（假设在 id=5 和 id=7 之间的间隙）

-- 事务 A
INSERT INTO users(id) VALUES(6);  
-- 需要等 B 释放间隙锁

-- 事务 B
INSERT INTO users(id) VALUES(5);  
-- 需要等 A 释放间隙锁
-- 死锁！

LOCK TABLES users READ;    -- 加表级读锁
LOCK TABLES users WRITE;   -- 加表级写锁
UNLOCK TABLES;             -- 释放锁

ALTER TABLE users ADD COLUMN age INT;
-- InnoDB 在 DDL 过程中需要使用表锁

-- name 没有索引的场景
DELETE FROM users WHERE name = 'Tom';
-- 可能锁住整张表（取决于 MySQL 版本和优化器决策）

SELECT * FROM articles WHERE MATCH(title) AGAINST('database');
-- 全文索引搜索需要表级锁

-- 会话 A：长查询
SELECT COUNT(*) FROM big_table;  -- 持有 MDL_SHARED_READ

-- 会话 B：DDL 操作
ALTER TABLE big_table ADD INDEX idx_name(name);  
-- 需要 MDL_EXCLUSIVE → 等待会话 A 释放

-- 会话 C：新的查询
SELECT * FROM big_table WHERE id = 1;
-- 需要 MDL_SHARED_READ → 被会话 B 的 MDL_EXCLUSIVE 阻塞

-- MySQL 8.0
SELECT * FROM performance_schema.metadata_locks;

-- 查看哪些语句在等待 MDL 锁
SELECT * FROM sys.schema_table_lock_waits;

-- 找到阻塞的线程并处理
SHOW PROCESSLIST;

-- 必要时 kill 阻塞连接
KILL [connection_id];

-- 1. 会话 A 开始长事务
BEGIN;
SELECT * FROM users WHERE id = 1;  -- 获得 MDL_SHARED_READ

-- 2. 会话 B 尝试 DDL
ALTER TABLE users ADD INDEX idx_name(name);  
-- 等待 MDL_EXCLUSIVE，被会话 A 阻塞

-- 3. 其他所有会话
SELECT * FROM users WHERE id = 2;  
-- 等待 MDL_SHARED_READ，被会话 B 阻塞
-- 全部挂起！服务不可用

-- 会话 A
LOCK TABLES users WRITE;  -- 加写表锁
UPDATE users SET name = 'Tom' WHERE id = 1;  -- ✓

-- 会话 B（被阻塞）
SELECT * FROM users WHERE id = 2;  -- 等待表锁
UPDATE users SET name = 'Jerry' WHERE id = 1;  -- 等待表锁

事务 A：对行加 X 锁

1. 先在表 users 上加 IX 锁（表级）
2. 再对行 id=1 加 X 锁（行级）

事务 B：对行加 S 锁

1. 先在表 users 上加 IS 锁（表级）   ← ✅ 兼容 IX
2. 再对行 id=1 加 S 锁（行级）      ← ❌ 与 X 锁互斥，等待

事务 C：锁整张表

尝试加表级 S 锁 → 检查表上是否有 IX 锁 → 有 → 互斥 → 等待
                    无需检查任何行级锁！

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id = 1 FOR UPDATE;
-- 自动：users 表上加 IX 锁，行上加 X 锁

-- 事务 B（另一个会话）
LOCK TABLES users WRITE;
-- 需要加 X 表锁
-- 检查到 users 表上有 IX 锁 → 互斥 → 等待事务 A 释放

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id = 1 FOR UPDATE;

-- 事务 C
LOCK TABLES users READ;
-- 需要加 S 表锁
-- 检查到 users 表上有 IX 锁 → 互斥 → 等待事务 A 释放

-- MySQL 8.0
SELECT * FROM performance_schema.data_locks;

-- 查看 LOCK_MODE 列：
-- LOCK_MODE = 'IS' 表示意向共享锁（表级）
-- LOCK_MODE = 'IX' 表示意向排他锁（表级）

事务 A 读取 id=1 的数据 → 加读锁
事务 B 要修改 id=1 的数据 → ❌ 被阻塞，等 A 释放锁

事务 A 读取 id=1 的数据 → 看到快照版本（不加锁）
事务 B 修改 id=1 的数据 → 创建新版本（也不阻塞写）
事务 A 后续再读 id=1 → 仍然看到旧版本（一致性不变）

MVCC 实现
│
├── 1. 隐藏列（Hidden Columns）
│   ├── DB_TRX_ID：最后修改该行的事务 ID
│   └── DB_ROLL_PTR：指向前一版本的 Undo Log 指针
│
├── 2. Undo Log（回滚日志）
│   └── 存储每个版本数据变更前的旧值，形成版本链
│
├── 3. Read View（读视图）
│   └── 决定当前事务能看到哪些版本的核心数据结构
│
└── 4. Purge 线程（清理线程）
    └── 定期清理不再被任何事务需要的旧版本数据

用户发起 SELECT
    │
    ▼
从聚簇索引获取当前版本
    │
    ▼
查看 DB_TRX_ID（该版本由哪个事务创建的）
    │
    ▼
与当前事务的 Read View 比较
    │
    ├── 可见 → 直接返回该版本
    │
    └── 不可见
            │
            ▼
        通过 DB_ROLL_PTR 找到 Undo Log 中的前一版本
            │
            ▼
        检查该版本的 DB_TRX_ID 对当前事务是否可见
            │
            ├── 可见 → 返回该版本
            │
            └── 不可见 → 继续沿版本链向前追溯

用户发起 UPDATE
    │
    ▼
对目标行加锁（防止并发修改）
    │
    ▼
将当前版本的旧值写入 Undo Log
    │
    ▼
更新行的数据为新值
    │
    ▼
更新 DB_TRX_ID 为当前事务 ID
    │
    ▼
更新 DB_ROLL_PTR 指向旧版本的 Undo Log
    │
    ▼
释放锁

SELECT 1 → 创建 Read View A（看到此刻已提交的数据）
SELECT 2 → 创建 Read View B（看到包括事务 T 提交后的数据）
         → 两次结果可能不同

SELECT 1 → 创建 Read View A（记录此刻活跃事务列表）
SELECT 2 → 复用 Read View A（和第一次看到的一样）
         → 两次结果一致

-- 事务 A 读取
SELECT * FROM user WHERE id = 1;  -- 加读锁，其他人不能写

-- 事务 B 无法写入
UPDATE user SET name = '李四' WHERE id = 1;  -- ❌ 阻塞！

-- 事务 A 读取（不加锁）
SELECT * FROM user WHERE id = 1;  -- 直接读快照，不用锁

-- 事务 B 修改（不阻塞）
UPDATE user SET name = '李四' WHERE id = 1;  -- ✅ 成功

-- 事务 A 再次读取（仍然看到旧版本，不会被 B 的修改影响）
SELECT * FROM user WHERE id = 1;  -- 看到事务 B 修改前的数据

-- 报表生成场景
BEGIN;
SELECT SUM(amount) FROM orders;  -- 读到包括未发货的订单
-- 另一个事务修改了订单状态
SELECT SUM(discount) FROM orders;  -- 基于已经修改后的数据
-- 总和计算使用了不同时间点的数据，结果不正确

BEGIN;
-- 第一次 SELECT 创建 Read View，记录此刻的所有活跃事务
SELECT SUM(amount) FROM orders;  -- 基于快照
SELECT SUM(discount) FROM orders;  -- 基于同一快照
-- 两个结果基于同一数据状态，逻辑一致
COMMIT;

START TRANSACTION;
UPDATE user SET name = 'A' WHERE id = 1;  -- 旧版本 'X' 存入 Undo
UPDATE user SET name = 'B' WHERE id = 2;  -- 旧版本 'Y' 存入 Undo

-- ...业务出错
ROLLBACK;
-- 从 Undo Log 中恢复：id=1 → 'X', id=2 → 'Y'

-- 需要统计某账户当日所有流水（不关心实时值）
BEGIN;
SELECT * FROM transaction WHERE account_id = 123;  -- 快照
-- 此时有另一个转账事务在修改 account=123
-- MVCC 保证两次 SELECT 结果一致
SELECT SUM(amount) FROM transaction WHERE account_id = 123;  -- 基于同一快照
COMMIT;

-- mysqldump 使用 --single-transaction 导出
-- 利用 MVCC 在 REPEATABLE READ 下获得一致性快照
-- 导出过程中其他事务可以继续写入，互不阻塞

MySQL 重启后，自增 ID 的恢复方式：
SELECT MAX(auto_inc_col) FROM t FOR UPDATE;

INSERT INTO users(name) VALUES('Tom');
-- 会话层级：获取 AUTO-INC 表锁 → 分配 ID → 释放 AUTO-INC 表锁
-- 对于所有 INSERT 都使用表级锁

-- 简答的 INSERT（行数可预知）
INSERT INTO users(name) VALUES('Tom');
INSERT INTO users(name) VALUES('Jerry'), ('Bob');
-- 使用轻量互斥量（mutex），性能很好

-- 批量插入（行数不可预知）
INSERT INTO users(name) SELECT name FROM old_users;
-- 使用表级自增锁，保证连续 ID

INSERT INTO users(name) VALUES('Tom');
INSERT INTO users(name) SELECT name FROM old_users;
-- 全部使用轻量互斥量
-- 不保证 ID 连续性

BEGIN;
INSERT INTO users(name) VALUES('Tom');
  -- 1. 获取自增锁（或互斥量）
  -- 2. 分配 ID=100
  -- 3. 释放自增锁 ← 此时锁已经释放了
  -- 4. 插入数据行（持有行锁）
COMMIT;

-- 原因 1：事务回滚
BEGIN;
INSERT INTO users(name) VALUES('Tom');  -- 分配 ID=100
ROLLBACK;  -- 行被回滚，但 ID=100 已被分配
INSERT INTO users(name) VALUES('Jerry');  -- ID=101，而不是 100

-- 原因 2：批量插入预留过多
INSERT INTO users(name) VALUES('A'),('B'),('C');  -- 预留 3 个 ID
-- 实际只用了 3 个，但如果预留发生在并发环境下，可能有些 ID 被浪费

-- 原因 3：MySQL 重启
-- MySQL 重启后，自增起始值可能改变

事务 A：INSERT INTO users(name) SELECT name FROM big_table;  -- 预留 100 个 ID
事务 B：INSERT INTO users(name) VALUES('New');  -- 得到 ID=101，而不是在 A 预留的 ID 之间

-- 主库
INSERT INTO t VALUES(NULL, 'a');  -- ID=1
INSERT INTO t VALUES(NULL, 'b');  -- ID=2
-- 由于交错，可能 ID 顺序不同

-- 查看当前模式
SELECT @@innodb_autoinc_lock_mode;
-- MySQL 8.0 默认：1（consecutive）

-- 设置模式（修改配置文件 my.cnf）
-- [mysqld]
-- innodb_autoinc_lock_mode = 2

-- 查看当前自增值
SELECT AUTO_INCREMENT FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'dbname' AND TABLE_NAME = 'users';

-- 修改自增值
ALTER TABLE users AUTO_INCREMENT = 1000;

-- 事务 A 持有 id=1 的 X 锁
UPDATE users SET name = 'A' WHERE id = 1;

-- 事务 B 不能同时持有 id=1 的 X 锁
UPDATE users SET name = 'B' WHERE id = 1;  -- 等待 A

-- 事务 A：持有 id=1 的锁，等待 id=2 的锁
UPDATE users SET balance = 100 WHERE id = 1;
UPDATE users SET balance = 200 WHERE id = 2;  -- 等待 B 释放

-- 事务 B：持有 id=2 的锁，等待 id=1 的锁
UPDATE users SET balance = 300 WHERE id = 2;
UPDATE users SET balance = 400 WHERE id = 1;  -- 等待 A 释放

-- 事务 A 持有 id=1 的锁
-- 无论事务 B 愿意付出什么代价，都不能让 A 交出锁
-- A 只能在 COMMIT 或 ROLLBACK 后释放

事务 A → 等待 → 事务 B → 等待 → 事务 C → 等待 → 事务 A
             ↑                               |
             └───────────────────────────────┘

-- 经典的两事务循环等待
-- 事务 A 持有 id=1，等待 id=2
-- 事务 B 持有 id=2，等待 id=1

-- 图表示：
-- A → (等待 id=2) → B
-- B → (等待 id=1) → A
-- A ↔ B 形成循环

三个事务的循环等待：
事务 A：持有 id=1，等待 id=2
事务 B：持有 id=2，等待 id=3  
事务 C：持有 id=3，等待 id=1

等待环：A → B → C → A

-- 初始数据
-- users 表中有 id=1 和 id=2 两条记录

-- 时间点 1：事务 A 执行
BEGIN;
UPDATE users SET balance = balance - 100 WHERE id = 1;  -- 锁 id=1

-- 时间点 2：事务 B 执行
BEGIN;
UPDATE users SET balance = balance - 200 WHERE id = 2;  -- 锁 id=2

-- 时间点 3：事务 A 执行
UPDATE users SET balance = balance + 100 WHERE id = 2;
-- 等待事务 B 释放 id=2 ← Hold and Wait

-- 时间点 4：事务 B 执行
UPDATE users SET balance = balance + 200 WHERE id = 1;
-- 等待事务 A 释放 id=1 ← 循环等待形成

-- 此时：死锁！InnoDB 会检测到并处理

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id = 5 FOR UPDATE;
-- id=5 不存在，锁间隙（假设已有 id=3, id=7）

-- 事务 B
BEGIN;
SELECT * FROM users WHERE id = 6 FOR UPDATE;
-- id=6 不存在，也锁在这个间隙（假设在 id=5 和 id=7 之间的间隙）

-- 事务 A
INSERT INTO users(id) VALUES(6);  
-- 需要等 B 释放间隙锁

-- 事务 B
INSERT INTO users(id) VALUES(5);  
-- 需要等 A 释放间隙锁
-- 死锁！

LOCK TABLES users READ;    -- 加表级读锁
LOCK TABLES users WRITE;   -- 加表级写锁
UNLOCK TABLES;             -- 释放锁

ALTER TABLE users ADD COLUMN age INT;
-- InnoDB 在 DDL 过程中需要使用表锁

-- name 没有索引的场景
DELETE FROM users WHERE name = 'Tom';
-- 可能锁住整张表（取决于 MySQL 版本和优化器决策）

SELECT * FROM articles WHERE MATCH(title) AGAINST('database');
-- 全文索引搜索需要表级锁

-- 会话 A：长查询
SELECT COUNT(*) FROM big_table;  -- 持有 MDL_SHARED_READ

-- 会话 B：DDL 操作
ALTER TABLE big_table ADD INDEX idx_name(name);  
-- 需要 MDL_EXCLUSIVE → 等待会话 A 释放

-- 会话 C：新的查询
SELECT * FROM big_table WHERE id = 1;
-- 需要 MDL_SHARED_READ → 被会话 B 的 MDL_EXCLUSIVE 阻塞

-- MySQL 8.0
SELECT * FROM performance_schema.metadata_locks;

-- 查看哪些语句在等待 MDL 锁
SELECT * FROM sys.schema_table_lock_waits;

-- 找到阻塞的线程并处理
SHOW PROCESSLIST;

-- 必要时 kill 阻塞连接
KILL [connection_id];

-- 1. 会话 A 开始长事务
BEGIN;
SELECT * FROM users WHERE id = 1;  -- 获得 MDL_SHARED_READ

-- 2. 会话 B 尝试 DDL
ALTER TABLE users ADD INDEX idx_name(name);  
-- 等待 MDL_EXCLUSIVE，被会话 A 阻塞

-- 3. 其他所有会话
SELECT * FROM users WHERE id = 2;  
-- 等待 MDL_SHARED_READ，被会话 B 阻塞
-- 全部挂起！服务不可用

-- 会话 A
LOCK TABLES users WRITE;  -- 加写表锁
UPDATE users SET name = 'Tom' WHERE id = 1;  -- ✓

-- 会话 B（被阻塞）
SELECT * FROM users WHERE id = 2;  -- 等待表锁
UPDATE users SET name = 'Jerry' WHERE id = 1;  -- 等待表锁

事务 A：对行加 X 锁

1. 先在表 users 上加 IX 锁（表级）
2. 再对行 id=1 加 X 锁（行级）

事务 B：对行加 S 锁

1. 先在表 users 上加 IS 锁（表级）   ← ✅ 兼容 IX
2. 再对行 id=1 加 S 锁（行级）      ← ❌ 与 X 锁互斥，等待

事务 C：锁整张表

尝试加表级 S 锁 → 检查表上是否有 IX 锁 → 有 → 互斥 → 等待
                    无需检查任何行级锁！

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id = 1 FOR UPDATE;
-- 自动：users 表上加 IX 锁，行上加 X 锁

-- 事务 B（另一个会话）
LOCK TABLES users WRITE;
-- 需要加 X 表锁
-- 检查到 users 表上有 IX 锁 → 互斥 → 等待事务 A 释放

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id = 1 FOR UPDATE;

-- 事务 C
LOCK TABLES users READ;
-- 需要加 S 表锁
-- 检查到 users 表上有 IX 锁 → 互斥 → 等待事务 A 释放

-- MySQL 8.0
SELECT * FROM performance_schema.data_locks;

-- 查看 LOCK_MODE 列：
-- LOCK_MODE = 'IS' 表示意向共享锁（表级）
-- LOCK_MODE = 'IX' 表示意向排他锁（表级）

WATCH key            -- 监视一个或多个 key
MULTI                -- 开启事务块
...                  -- 执行命令
EXEC                 -- 执行事务，如果 WATCH 的 key 被修改则放弃执行
UNWATCH              -- 取消监视

-- 伪代码：用 WATCH 实现乐观锁
WATCH balance
balance_val = GET balance  -- 获取当前余额
if balance_val >= amount
    MULTI
    DECRBY balance amount
    EXEC                  -- 如果在 WATCH 之后 balance 被其他客户端修改，EXEC 返回 nil
end

# Python 实现：基于 WATCH 的库存扣减
def decrease_stock(redis, product_id, count):
    key = f"stock:{product_id}"
    while True:
        redis.watch(key)
        stock = redis.get(key)
        if stock is None or int(stock) < count:
            redis.unwatch()
            return False
        pipe = redis.pipeline(transaction=True)
        pipe.decrby(key, count)
        result = pipe.execute()
        if result is not None:
            return True
    # 执行失败自动重试

WATCH key1 key2 key3

redis> WATCH stock:1001          ← 监视库存 key
OK
redis> GET stock:1001            ← 查看当前库存
"10"
redis> MULTI                     ← 开启事务
OK
redis> DECR stock:1001           ← 扣减库存（入队）
QUEUED
redis> EXEC                      ← 执行事务
1) (integer) 9                   ← 执行成功

# 客户端 1
redis> WATCH stock:1001          ← 监视库存
OK
redis> GET stock:1001
"10"
redis> MULTI
OK
redis> DECR stock:1001           ← 在 EXEC 之前...

                                # 客户端 2 修改了 stock:1001
                                redis> DECR stock:1001
                                (integer) 9

redis> EXEC
(nil)                          ← 事务执行失败！因为 key 被修改了

客户端 A                             客户端 B
  │                                    │
  ├─ WATCH stock:1001                  │
  ├─ GET stock:1001 → 10               │
  │                                    │
  ├─ MULTI                             │
  │                                    ├─ DECR stock:1001 → 9
  ├─ DECR stock:1001 (入队)            │
  │                                    │
  ├─ EXEC                              │
  │   → (nil)                          │
  │  因为 key 被修改了！                │
  │  事务自动放弃                       │

public boolean deductStock(Long productId, int quantity) {
    String key = "stock:" + productId;

    // 使用 WATCH + 事务实现安全扣减
    return redisTemplate.execute(new SessionCallback<Boolean>() {
        @Override
        public Boolean execute(RedisOperations ops) {
            while (true) {  // 重试循环
                ops.watch(key);

                // 获取当前库存
                Integer stock = (Integer) ops.opsForValue().get(key);
                if (stock == null || stock < quantity) {
                    ops.unwatch();
                    return false;  // 库存不足
                }

                ops.multi();
                ops.opsForValue().decrement(key, quantity);

                List<Object> results = ops.exec();
                if (results != null && !results.isEmpty()) {
                    return true;  // CAS 成功！
                }
                // CAS 失败（被其他客户端修改了），重试
            }
        }
    });
}

// 多操作组合的原子性保证
public boolean incrementUserStats(String userId) {
    String loginKey = "stats:" + userId + ":logins";
    String pointsKey = "stats:" + userId + ":points";

    return redisTemplate.execute(new SessionCallback<Boolean>() {
        @Override
        public Boolean execute(RedisOperations ops) {
            ops.watch(loginKey, pointsKey);

            ops.multi();
            ops.opsForValue().increment(loginKey);  // 登录次数 +1
            ops.opsForValue().increment(pointsKey, 5);  // 积分 +5

            List<Object> results = ops.exec();
            return results != null && !results.isEmpty();
        }
    });
}

redis> WATCH key1 key2    ← 监视 key1 和 key2
OK
redis> EXEC               ← 执行后，WATCH 自动取消
# 或者
redis> UNWATCH            ← 手动取消所有监视
OK

WATCH key1 key2 ...    ① 监视
GET key1 / GET key2    ② 读取当前值（业务判断基准）
MULTI                   ③ 开启事务
命令1                   ④ 入队
命令2
EXEC                    ⑤ 执行
                        ⑥ 如果 WATCH key 被修改 → 返回 nil
                        ⑦ 如果没被修改 → 执行成功，返回结果
                        ⑧ 自动 UNWATCH

1. 在 ZK 的 /locks/ 下创建临时顺序节点 /locks/lock-000000001
2. 获取 /locks/ 下所有子节点，排序
3. 判断自己的节点序号是否最小
   是 → 获得锁
   否 → 监听前一个节点的删除事件
4. 前一个节点删除 → 重复步骤 2-3

@Bean
public CuratorFramework curatorFramework() {
    return CuratorFrameworkFactory.newClient(
        "zk1:2181,zk2:2181,zk3:2181",
        new ExponentialBackoffRetry(1000, 3)
    );
}

public void processWithZkLock() {
    InterProcessMutex lock = new InterProcessMutex(client, "/locks/order");

    try {
        if (lock.acquire(10, TimeUnit.SECONDS)) {
            try {
                processOrder();
            } finally {
                lock.release();
            }
        }
    } catch (Exception e) {
        // 异常处理
    }
}

// Go 语言示例（etcd 官方 concurrency 包）
import "go.etcd.io/etcd/client/v3/concurrency"

func acquireLock() error {
    session, _ := concurrency.NewSession(cli)
    // Session 自动续约 lease
    mutex := concurrency.NewMutex(session, "/mylock/")
    // 尝试加锁，超时 5 秒
    return mutex.Lock(context.TODO())
}

CREATE TABLE distributed_lock (
    lock_key VARCHAR(64) PRIMARY KEY,
    holder_id VARCHAR(64) NOT NULL,
    expire_at DATETIME NOT NULL,
    version INT NOT NULL DEFAULT 0
);

-- 加锁：插入成功即获得锁
INSERT INTO distributed_lock (lock_key, holder_id, expire_at)
VALUES ('order_lock_1001', 'client_A', NOW() + INTERVAL 30 SECOND);
-- 唯一索引冲突 → 插入失败 → 加锁失败

-- 悲观锁
BEGIN;
SELECT * FROM distributed_lock 
WHERE lock_key = 'order_lock_1001' AND holder_id = 'client_A'
FOR UPDATE;
UPDATE resource SET status = 'processing' WHERE order_id = 1001;
COMMIT;

add lock_key request_id 30  ← 类似 SETNX+EXPIRE，原子操作

// 决策树
if (已部署 ZK/Etcd) {
    // 直接使用已有的协调组件
    return ZooKeeperCurator / EtcdLock;
} else if (一致性要求极高) {
    // 金融、交易场景
    return EtcdLock;          // 云原生首选
    // 或 ZooKeeperCurator;  // 传统架构首选
} else if (已有 Redis) {
    if (锁丢失可接受 || 有补偿机制) {
        return RedissonLock;  // 足够
    } else {
        return Redlock;       // 多节点
    }
} else {
    // 没有分布式协调中间件
    return DataBaseLock;      // 临时的简单方案
}

// 锁 TTL = 5 秒
lock(key, value, 5, TimeUnit.SECONDS);

// 业务执行时间取决于很多因素：
// - 数据库查询速度（可能慢查询）
// - 外部 API 调用（可能超时重试）
// - GC pause（Stop-The-World）
// - 网络波动（RPC 延迟）
doBusiness();  // 有时 100ms，有时 5 秒+

时间线：

0s:   加锁成功（TTL=5s）
1-4s: 执行业务逻辑...
5s:   锁自动过期 ❗（业务还在执行！）
5s:   客户端 B 获取到同一把锁
6s:   客户端 A 业务完成，释放锁（释放了 B 的锁）
6s:   客户端 A 和 B 同时操作同一资源 → 并发异常

public class LockRenewalService {
    private final StringRedisTemplate redis;
    private final ScheduledExecutorService scheduler;
    private final Map<String, ScheduledFuture> renewalTasks = new ConcurrentHashMap<>();

    public void startRenewal(String key, String holder, int ttlSeconds) {
        ScheduledFuture task = scheduler.scheduleAtFixedRate(() -> {
            // 续约 Lua 脚本：检查持有者后续约
            String lua = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                         "  return redis.call('expire', KEYS[1], ARGV[2]) " +
                         "else return 0 end";
            redis.execute(new DefaultRedisScript<>(lua, Long.class),
                Collections.singletonList(key), holder, String.valueOf(ttlSeconds));
        }, ttlSeconds / 3, ttlSeconds / 3, TimeUnit.SECONDS);

        renewalTasks.put(key, task);
    }

    public void stopRenewal(String key) {
        ScheduledFuture task = renewalTasks.remove(key);
        if (task != null) {
            task.cancel(false);
        }
    }
}

public class ConditionalRenewal {
    private volatile boolean businessCompleted = false;

    public void executeWithRenewal(String key, String holder, int ttlSeconds) {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

        scheduler.scheduleAtFixedRate(() -> {
            if (businessCompleted) {
                return;  // 业务已完成，不续约
            }
            // 续约
            renewLock(key, holder, ttlSeconds);
        }, ttlSeconds / 3, ttlSeconds / 3, TimeUnit.SECONDS);

        try {
            doBusiness();
        } finally {
            businessCompleted = true;
            scheduler.shutdown();
            unlock(key, holder);
        }
    }
}