MySQL事务与锁：并发控制的艺术

/**
 * 秒杀服务（无并发控制）
 */
@Service
public class SeckillService {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private OrderMapper orderMapper;

    /**
     * 秒杀下单（存在并发问题）
     */
    public boolean seckill(Long productId, Long userId) {
        // 1. 读取库存
        Product product = productMapper.selectById(productId);
        int stock = product.getStock();

        // 2. 判断库存是否足够
        if (stock <= 0) {
            return false;  // 库存不足
        }

        // 3. 扣减库存
        product.setStock(stock - 1);
        productMapper.updateById(product);

        // 4. 创建订单
        Order order = new Order();
        order.setUserId(userId);
        order.setProductId(productId);
        order.setAmount(1.00);  // 秒杀价1元
        orderMapper.insert(order);

        return true;
    }
}

public class SeckillServiceTest {

    @Test
    public void testConcurrency() throws Exception {
        SeckillService service = new SeckillService();
        Long productId = 1L;  // iPhone库存10件

        // 模拟1000个用户并发抢购
        CountDownLatch latch = new CountDownLatch(1000);
        AtomicInteger successCount = new AtomicInteger(0);

        for (int i = 0; i < 1000; i++) {
            final int userId = i;
            new Thread(() -> {
                try {
                    latch.countDown();
                    latch.await();  // 所有线程同时开始

                    boolean success = service.seckill(productId, (long) userId);
                    if (success) {
                        successCount.incrementAndGet();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }

        Thread.sleep(5000);  // 等待所有线程完成

        System.out.println("成功下单人数: " + successCount.get());
        System.out.println("最终库存: " + productMapper.selectById(productId).getStock());

        // 预期结果:
        // 成功下单人数: 10
        // 最终库存: 0

        // 实际结果:
        // 成功下单人数: 856  ❌ 超卖846件
        // 最终库存: -846     ❌ 库存为负数
    }
}

/**
 * 秒杀服务（悲观锁）
 */
@Service
public class SeckillServiceWithPessimisticLock {

    @Autowired
    private ProductMapper productMapper;

    /**
     * 秒杀下单（使用FOR UPDATE加锁）
     */
    @Transactional
    public boolean seckill(Long productId, Long userId) {
        // 1. 加排他锁读取库存
        Product product = productMapper.selectByIdForUpdate(productId);
        int stock = product.getStock();

        // 2. 判断库存
        if (stock <= 0) {
            return false;
        }

        // 3. 扣减库存
        product.setStock(stock - 1);
        productMapper.updateById(product);

        // 4. 创建订单
        Order order = new Order();
        order.setUserId(userId);
        order.setProductId(productId);
        order.setAmount(1.00);
        orderMapper.insert(order);

        return true;
    }
}

-- ProductMapper.selectByIdForUpdate()
BEGIN;

-- 加排他锁（其他事务无法读写）
SELECT * FROM products WHERE id = 1 FOR UPDATE;
-- 此时库存=10

-- 扣减库存
UPDATE products SET stock = stock - 1 WHERE id = 1;

COMMIT;  -- 释放锁

@Test
public void testPessimisticLock() throws Exception {
    // 1000个并发请求
    long start = System.currentTimeMillis();

    // ... 并发测试代码 ...

    long duration = System.currentTimeMillis() - start;
    System.out.println("总耗时: " + duration + "ms");
    // 结果: 约5000ms

    // 分析:
    // 1000个请求串行执行
    // 每个请求约5ms
    // 总耗时 = 1000 × 5ms = 5000ms
}

-- 事务A
BEGIN;
SELECT * FROM products WHERE id = 1 FOR UPDATE;  -- 锁住商品1
SELECT * FROM products WHERE id = 2 FOR UPDATE;  -- 等待商品2的锁

-- 事务B
BEGIN;
SELECT * FROM products WHERE id = 2 FOR UPDATE;  -- 锁住商品2
SELECT * FROM products WHERE id = 1 FOR UPDATE;  -- 等待商品1的锁

-- 死锁! A等B,B等A

-- 增加version字段
ALTER TABLE products ADD COLUMN version INT NOT NULL DEFAULT 0;

/**
 * 秒杀服务（乐观锁）
 */
@Service
public class SeckillServiceWithOptimisticLock {

    /**
     * 秒杀下单（乐观锁 + 重试）
     */
    public boolean seckill(Long productId, Long userId) {
        int maxRetry = 3;  // 最大重试次数

        for (int i = 0; i < maxRetry; i++) {
            // 1. 读取库存和版本号
            Product product = productMapper.selectById(productId);
            int stock = product.getStock();
            int version = product.getVersion();

            // 2. 判断库存
            if (stock <= 0) {
                return false;
            }

            // 3. 乐观锁更新（检查版本号）
            int updatedRows = productMapper.updateStockWithVersion(
                productId,
                stock - 1,
                version + 1,
                version  // WHERE version = ?
            );

            // 4. 更新成功,创建订单
            if (updatedRows > 0) {
                Order order = new Order();
                order.setUserId(userId);
                order.setProductId(productId);
                order.setAmount(1.00);
                orderMapper.insert(order);
                return true;
            }

            // 5. 更新失败,重试
            System.out.println("版本冲突,重试第" + (i + 1) + "次");
        }

        return false;  // 重试3次仍失败
    }
}

-- ProductMapper.updateStockWithVersion()
UPDATE products
SET stock = ?, version = ?
WHERE id = ? AND version = ?;

-- 执行流程:
-- 事务A: UPDATE ... WHERE version = 5  → 成功,version变为6,affected_rows=1
-- 事务B: UPDATE ... WHERE version = 5  → 失败,version已经是6,affected_rows=0
-- 事务B: 重试,读取version=6,再次UPDATE → 成功

@Test
public void testOptimisticLock() throws Exception {
    // 1000个并发请求
    long start = System.currentTimeMillis();

    AtomicInteger retryCount = new AtomicInteger(0);

    // ... 并发测试代码 ...

    long duration = System.currentTimeMillis() - start;
    System.out.println("总耗时: " + duration + "ms");
    System.out.println("总重试次数: " + retryCount.get());

    // 结果:
    // 总耗时: 800ms  ✅ 比悲观锁快6倍
    // 总重试次数: 2456  ⚠️ 平均每个请求重试2.5次
}

-- 每行数据有3个隐藏列
CREATE TABLE products (
    id BIGINT PRIMARY KEY,
    stock INT,
    -- 隐藏列（MySQL自动维护）
    DB_TRX_ID BIGINT,      -- 最后修改该行的事务ID
    DB_ROLL_PTR BIGINT,    -- 指向undo log中的历史版本
    DB_ROW_ID BIGINT       -- 行ID（如果没有主键）
);

-- 快照读（不加锁,读历史版本）
SELECT * FROM products WHERE id = 1;

-- 当前读（加锁,读最新版本）
SELECT * FROM products WHERE id = 1 FOR UPDATE;       -- 加写锁
SELECT * FROM products WHERE id = 1 LOCK IN SHARE MODE;  -- 加读锁
UPDATE products SET stock = stock - 1 WHERE id = 1;   -- 加写锁
DELETE FROM products WHERE id = 1;                    -- 加写锁

/**
 * 秒杀服务（MVCC）
 */
@Service
public class SeckillServiceWithMVCC {

    /**
     * 秒杀下单（利用MVCC的当前读）
     */
    @Transactional(isolation = Isolation.REPEATABLE_READ)
    public boolean seckill(Long productId, Long userId) {
        // 1. 当前读（加锁,读最新版本）
        Product product = productMapper.selectByIdForUpdate(productId);
        int stock = product.getStock();

        // 2. 判断库存
        if (stock <= 0) {
            return false;
        }

        // 3. 扣减库存（原子操作）
        int updatedRows = productMapper.decreaseStock(productId, 1);
        if (updatedRows == 0) {
            return false;
        }

        // 4. 创建订单
        Order order = new Order();
        order.setUserId(userId);
        order.setProductId(productId);
        order.setAmount(1.00);
        orderMapper.insert(order);

        return true;
    }
}

-- 方案1: 直接在UPDATE中判断库存
UPDATE products
SET stock = stock - 1
WHERE id = ? AND stock > 0;

-- 如果affected_rows=0,说明库存不足
-- 无需SELECT FOR UPDATE,减少锁等待

-- 方案2: 使用CAS（Compare And Swap）
UPDATE products
SET stock = stock - 1
WHERE id = ? AND stock = ?;

-- 需要先读取库存,然后在UPDATE中判断库存是否变化

@Test
public void testMVCC() throws Exception {
    // 1000个并发请求
    long start = System.currentTimeMillis();

    // ... 并发测试代码 ...

    long duration = System.currentTimeMillis() - start;
    System.out.println("总耗时: " + duration + "ms");

    // 结果:
    // 方案1（SELECT FOR UPDATE + UPDATE）: 约1000ms
    // 方案2（直接UPDATE）: 约200ms  ✅ 性能最好

    // 分析:
    // 方案2无需SELECT,减少了50%的SQL执行
    // 无需加行锁,其他事务可以并发执行
}

-- 初始状态
stock=13, DB_TRX_ID=97

-- 事务98修改
BEGIN;  -- 事务ID=98
UPDATE products SET stock = 12 WHERE id = 1;
COMMIT;

-- 生成undo log:
-- 当前版本: stock=12, DB_TRX_ID=98, DB_ROLL_PTR→版本1
-- 版本1: stock=13, DB_TRX_ID=97

-- 事务99修改
BEGIN;  -- 事务ID=99
UPDATE products SET stock = 11 WHERE id = 1;
COMMIT;

-- 生成undo log:
-- 当前版本: stock=11, DB_TRX_ID=99, DB_ROLL_PTR→版本1
-- 版本1: stock=12, DB_TRX_ID=98, DB_ROLL_PTR→版本2
-- 版本2: stock=13, DB_TRX_ID=97

/**
 * MVCC可见性判断（简化版）
 */
public class MVCCVisibility {

    static class ReadView {
        List<Long> m_ids;        // 活跃事务ID列表
        long min_trx_id;         // 最小活跃事务ID
        long max_trx_id;         // 下一个事务ID
        long creator_trx_id;     // 当前事务ID
    }

    /**
     * 判断版本是否可见
     */
    public static boolean isVisible(long trx_id, ReadView view) {
        // 规则1: 如果是当前事务的修改,可见
        if (trx_id == view.creator_trx_id) {
            return true;
        }

        // 规则2: 如果版本在ReadView生成前已提交,可见
        if (trx_id < view.min_trx_id) {
            return true;
        }

        // 规则3: 如果版本在ReadView生成后才开始,不可见
        if (trx_id >= view.max_trx_id) {
            return false;
        }

        // 规则4: 如果版本在活跃事务列表中,不可见
        if (view.m_ids.contains(trx_id)) {
            return false;
        }

        // 规则5: 其他情况,已提交,可见
        return true;
    }

    /**
     * 查找可见版本
     */
    public static Row findVisibleVersion(Row current, ReadView view) {
        Row version = current;

        // 沿着版本链查找
        while (version != null) {
            if (isVisible(version.DB_TRX_ID, view)) {
                return version;  // 找到可见版本
            }
            // 沿着DB_ROLL_PTR找下一个版本
            version = version.nextVersion;
        }

        return null;  // 无可见版本
    }
}

-- 当前读会加Next-Key Lock,锁住范围和间隙
SELECT * FROM orders WHERE user_id BETWEEN 100 AND 200 FOR UPDATE;

锁定范围:
  ├─ 记录锁:锁住user_id=100~200的所有行
  └─ 间隙锁:锁住(100, 200)之间的间隙

效果:
  其他事务无法在这个范围内插入新行
  避免幻读

-- 加表级读锁
LOCK TABLES products READ;
SELECT * FROM products;  -- 允许
UPDATE products SET stock = 10;  -- 报错:表已加读锁
UNLOCK TABLES;

-- 加表级写锁
LOCK TABLES products WRITE;
SELECT * FROM products;  -- 允许
UPDATE products SET stock = 10;  -- 允许
-- 其他事务无法读写（阻塞）
UNLOCK TABLES;

特点:
  ✅ 实现简单
  ❌ 粒度太大,并发性能差
  ❌ 容易死锁

-- MDL锁自动加,用于保护表结构

事务A:
  BEGIN;
  SELECT * FROM products;  -- 自动加MDL读锁

事务B:
  ALTER TABLE products ADD COLUMN name VARCHAR(50);  -- 等待MDL写锁,阻塞

事务A:
  COMMIT;  -- 释放MDL读锁

事务B:
  -- 获取MDL写锁,执行ALTER TABLE

作用:
  防止DDL和DML并发执行导致数据不一致

意向锁的作用:
  表级锁和行级锁的协调机制

问题:
  事务A对某一行加了X锁
  事务B想对整个表加X锁
  如何快速判断是否冲突?

传统方案:
  遍历所有行,检查是否有行锁
  性能差:O(n)

意向锁方案:
  事务A加行锁前,先加表级意向锁
  事务B检查表级锁时,只需检查意向锁
  性能好:O(1)

意向锁类型:
  IS锁（Intention Shared Lock）:表示事务打算加S行锁
  IX锁（Intention Exclusive Lock）:表示事务打算加X行锁

兼容性:
  IS与IS、IX、S兼容
  IX与IS、IX兼容
  S与IS兼容
  X与所有锁不兼容

-- 锁住索引记录
SELECT * FROM products WHERE id = 1 FOR UPDATE;

锁定范围:
  只锁id=1这一行的索引记录

示例:
┌────────────────────────────────────┐
│ 主键索引                            │
├────────────────────────────────────┤
│ id=1  [X锁] ← 被锁住                │
│ id=2                                │
│ id=3                                │
└────────────────────────────────────┘

其他事务:
  SELECT * FROM products WHERE id = 2;  -- 允许
  UPDATE products SET stock = 10 WHERE id = 1;  -- 阻塞

-- 锁住索引之间的间隙,防止插入
SELECT * FROM products WHERE id BETWEEN 10 AND 20 FOR UPDATE;

锁定范围（假设存在id=10,15,20）:
  (10, 15)  -- 间隙1
  (15, 20)  -- 间隙2

示例:
┌────────────────────────────────────┐
│ 主键索引                            │
├────────────────────────────────────┤
│ id=10 [X锁]                         │
│   ↓ 间隙1 [Gap锁] ← 被锁住          │
│ id=15 [X锁]                         │
│   ↓ 间隙2 [Gap锁] ← 被锁住          │
│ id=20 [X锁]                         │
└────────────────────────────────────┘

其他事务:
  INSERT INTO products (id, stock) VALUES (12, 100);  -- 阻塞（在间隙1中）
  INSERT INTO products (id, stock) VALUES (5, 100);   -- 允许（不在间隙中）
  UPDATE products SET stock = 10 WHERE id = 10;  -- 阻塞（记录被锁）

作用:
  防止幻读

-- 记录锁 + 间隙锁
SELECT * FROM products WHERE id >= 10 AND id <= 20 FOR UPDATE;

锁定范围:
  (5, 10]   -- Next-Key Lock 1
  (10, 15]  -- Next-Key Lock 2
  (15, 20]  -- Next-Key Lock 3
  (20, 25)  -- 间隙锁

InnoDB默认使用Next-Key Lock,既锁住记录,又锁住间隙

示例:
┌────────────────────────────────────┐
│ 主键索引                            │
├────────────────────────────────────┤
│ id=5                                │
│   ↓ (5, 10] [Next-Key锁]           │
│ id=10 [X锁]                         │
│   ↓ (10, 15] [Next-Key锁]          │
│ id=15 [X锁]                         │
│   ↓ (15, 20] [Next-Key锁]          │
│ id=20 [X锁]                         │
│   ↓ (20, 25) [Gap锁]               │
│ id=25                               │
└────────────────────────────────────┘

其他事务:
  INSERT INTO products (id, stock) VALUES (8, 100);   -- 阻塞
  INSERT INTO products (id, stock) VALUES (12, 100);  -- 阻塞
  INSERT INTO products (id, stock) VALUES (22, 100);  -- 阻塞
  INSERT INTO products (id, stock) VALUES (30, 100);  -- 允许

-- 场景1:唯一索引等值查询（记录存在）
SELECT * FROM products WHERE id = 10 FOR UPDATE;

锁:
  退化为记录锁（只锁id=10,不锁间隙）

-- 场景2:唯一索引等值查询（记录不存在）
SELECT * FROM products WHERE id = 12 FOR UPDATE;

锁:
  退化为间隙锁（只锁(10, 15),不锁记录）

-- 场景3:非唯一索引
SELECT * FROM products WHERE stock = 100 FOR UPDATE;

锁:
  使用Next-Key Lock（锁住所有stock=100的记录及其间隙）

-- 查看最近一次死锁
SHOW ENGINE INNODB STATUS\G

输出:
------------------------
LATEST DETECTED DEADLOCK
------------------------
2024-11-03 16:00:00 0x7f8e8c001700
*** (1) TRANSACTION:
TRANSACTION 12345, ACTIVE 10 sec starting index read
mysql tables in use 1, locked 1
LOCK WAIT 2 lock struct(s), heap size 1136, 1 row lock(s)
MySQL thread id 1, OS thread handle 123456, query id 100 localhost root updating
UPDATE products SET stock = 10 WHERE id = 2

*** (1) HOLDS THE LOCK(S):
RECORD LOCKS space id 0 page no 3 n bits 72 index PRIMARY of table `test`.`products` trx id 12345 lock_mode X locks rec but not gap
Record lock, heap no 2 PHYSICAL RECORD: n_fields 5; compact format; info bits 0
 0: len 8; hex 0000000000000001; asc         ;;  -- id=1

*** (1) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 0 page no 3 n bits 72 index PRIMARY of table `test`.`products` trx id 12345 lock_mode X locks rec but not gap waiting
Record lock, heap no 3 PHYSICAL RECORD: n_fields 5; compact format; info bits 0
 0: len 8; hex 0000000000000002; asc         ;;  -- id=2

*** (2) TRANSACTION:
TRANSACTION 12346, ACTIVE 8 sec starting index read
mysql tables in use 1, locked 1
3 lock struct(s), heap size 1136, 2 row lock(s)
MySQL thread id 2, OS thread handle 789012, query id 101 localhost root updating
UPDATE products SET stock = 10 WHERE id = 1

*** (2) HOLDS THE LOCK(S):
RECORD LOCKS space id 0 page no 3 n bits 72 index PRIMARY of table `test`.`products` trx id 12346 lock_mode X locks rec but not gap
Record lock, heap no 3 PHYSICAL RECORD: n_fields 5; compact format; info bits 0
 0: len 8; hex 0000000000000002; asc         ;;  -- id=2

*** (2) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 0 page no 3 n bits 72 index PRIMARY of table `test`.`products` trx id 12346 lock_mode X locks rec but not gap waiting
Record lock, heap no 2 PHYSICAL RECORD: n_fields 5; compact format; info bits 0
 0: len 8; hex 0000000000000001; asc         ;;  -- id=1

*** WE ROLL BACK TRANSACTION (1)  -- 回滚事务1

-- 转账事务
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;  -- A-100
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;  -- B+100
COMMIT;

undo log记录:
  undo log 1: UPDATE accounts SET balance = balance + 100 WHERE user_id = 1;  -- 反向操作
  undo log 2: UPDATE accounts SET balance = balance - 100 WHERE user_id = 2;  -- 反向操作

如果事务失败:
  ROLLBACK;
  → 执行undo log 2（B账户-100）
  → 执行undo log 1（A账户+100）
  → 数据回滚到事务开始前的状态

-- 约束保证
CREATE TABLE accounts (
    user_id BIGINT PRIMARY KEY,
    balance DECIMAL(10,2) NOT NULL CHECK (balance >= 0),  -- 余额不能为负
    CONSTRAINT chk_balance CHECK (balance >= 0)
);

-- 事务逻辑保证
BEGIN;
-- 检查A账户余额
SELECT balance FROM accounts WHERE user_id = 1;  -- 余额=100

-- 转账金额不能超过余额
IF balance < 100 THEN
    ROLLBACK;  -- 回滚
END IF;

-- 转账
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;

COMMIT;

一致性约束:
  ├─ 转账前:A余额=100, B余额=50, 总额=150
  ├─ 转账中:A余额=0, B余额=150, 总额=150
  └─ 转账后:A余额=0, B余额=150, 总额=150
  总额始终=150（一致性）

-- 转账事务
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;
COMMIT;  -- 提交

提交流程:
  1. 修改Buffer Pool中的数据页（内存）
  2. 写redo log到磁盘（顺序写,快）
  3. 返回客户端:提交成功
  4. 后台异步刷脏页到数据文件（随机写,慢）

如果在第4步之前崩溃:
  1. 重启MySQL
  2. 读取redo log
  3. 重放redo log中的操作
  4. 恢复到崩溃前的状态

@Transactional
public boolean seckill(Long productId, Long userId) {
    // 加排他锁
    Product product = productMapper.selectByIdForUpdate(productId);
    if (product.getStock() <= 0) {
        return false;
    }

    // 扣减库存
    product.setStock(product.getStock() - 1);
    productMapper.updateById(product);

    // 创建订单
    orderMapper.insert(order);
    return true;
}

性能测试:
  并发:1000 QPS
  响应时间:500ms
  问题:读写互斥,性能差

public boolean seckill(Long productId, Long userId) {
    // 直接扣减库存
    int updatedRows = productMapper.decreaseStock(productId, 1);
    if (updatedRows == 0) {
        return false;
    }

    // 创建订单
    orderMapper.insert(order);
    return true;
}

// SQL
UPDATE products
SET stock = stock - 1
WHERE id = ? AND stock > 0;

性能测试:
  并发:15000 QPS  ✅
  响应时间:50ms   ✅
  问题:无

public boolean seckill(Long productId, Long userId) {
    String key = "stock:" + productId;

    // 1. Redis原子扣减库存
    Long stock = redisTemplate.opsForValue().decrement(key);
    if (stock < 0) {
        // 库存不足,恢复库存
        redisTemplate.opsForValue().increment(key);
        return false;
    }

    // 2. 异步创建订单（发送MQ消息）
    OrderMessage message = new OrderMessage(userId, productId);
    rabbitTemplate.convertAndSend("order.queue", message);

    return true;
}

// MQ消费者（异步扣减MySQL库存）
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage message) {
    // 扣减MySQL库存
    productMapper.decreaseStock(message.getProductId(), 1);

    // 创建订单
    orderMapper.insert(order);
}

性能测试:
  并发:50000 QPS   ✅✅✅
  响应时间:10ms    ✅✅✅
  问题:需要Redis和MQ