锁冲突与死锁案例

-- 事务 A
BEGIN;
SELECT * FROM orders WHERE id = 1001 LOCK IN SHARE MODE;
-- 此时可以读取,但不能修改

-- 事务 B(同时执行)
BEGIN;
SELECT * FROM orders WHERE id = 1001 LOCK IN SHARE MODE; -- 成功,共享锁兼容
UPDATE orders SET status = 'PAID' WHERE id = 1001; -- 阻塞,需要排他锁
COMMIT;

-- 事务 A
BEGIN;
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
-- 此时其他事务不能读也不能写该行

-- 事务 B(同时执行)
BEGIN;
SELECT * FROM orders WHERE id = 1001 FOR UPDATE; -- 阻塞,等待 A 释放锁
COMMIT;

-- 表结构
CREATE TABLE users (
    id INT PRIMARY KEY,
    name VARCHAR(50),
    age INT,
    INDEX idx_age (age)
);

-- 情况1:通过主键更新,只锁定一行
UPDATE users SET name = 'Alice' WHERE id = 1; -- 行锁

-- 情况2:通过索引更新,锁定索引记录
UPDATE users SET name = 'Bob' WHERE age = 25; -- 行锁(锁定满足条件的索引项)

-- 情况3:无索引更新,锁住所有扫描到的行
UPDATE users SET name = 'Charlie' WHERE name = 'David'; -- name 列无索引,全表扫描,近似锁全表

-- 表数据:id = 1, 5, 10

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id > 3 AND id < 8 FOR UPDATE;
-- 锁定间隙 (1, 5) 和 (5, 10),以及 id=5 这一行

-- 事务 B(同时执行)
BEGIN;
INSERT INTO users VALUES(2, 'test'); -- 阻塞,因为 id=2 在间隙 (1, 5) 内
INSERT INTO users VALUES(8, 'test'); -- 阻塞,因为 id=8 在间隙 (5, 10) 内
COMMIT;

-- 表数据:id = 1, 5, 10

-- 事务 A
BEGIN;
SELECT * FROM users WHERE id <= 5 FOR UPDATE;
-- Next-Key Lock 锁定:(-∞, 1]、(1, 5]
-- 包括:间隙 (-∞, 1)、行 1、间隙 (1, 5)、行 5

-- 事务 B
BEGIN;
INSERT INTO users VALUES(0, 'test'); -- 阻塞
INSERT INTO users VALUES(2, 'test'); -- 阻塞
INSERT INTO users VALUES(5, 'test'); -- 阻塞(行锁冲突)
INSERT INTO users VALUES(10, 'test'); -- 成功
COMMIT;

-- 事务 A
BEGIN;
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
-- 自动获取:意向排他锁(表级) + 排他行锁(行级)

-- 事务 B(同时执行)
BEGIN;
LOCK TABLE orders WRITE; -- 阻塞,因为表有意向排他锁
COMMIT;

-- 不同隔离级别下的锁行为示例

-- 读已提交(RC)
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
BEGIN;
SELECT * FROM users WHERE id > 5 FOR UPDATE;
-- 只锁定 id > 5 的已存在记录(id=10)
-- 其他事务可以在 (5, 10) 和 (10, +∞) 插入新记录

-- 可重复读(RR)
SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
BEGIN;
SELECT * FROM users WHERE id > 5 FOR UPDATE;
-- 锁定 (5, +∞) 的所有间隙和记录
-- 其他事务不能在 (5, +∞) 范围内插入新记录

-- 查看锁等待超时时间(默认 50 秒)
SHOW VARIABLES LIKE 'innodb_lock_wait_timeout';

-- 设置锁等待超时时间
SET innodb_lock_wait_timeout = 30; -- 30 秒

-- 超时后报错
ERROR 1205 (HY000): Lock wait timeout exceeded; try restarting transaction

事务 A: 持有锁(资源 1) → 等待锁(资源 2)
                              ↑
                              |
事务 B: 持有锁(资源 2) ← 等待锁(资源 1)

-- 查看死锁检测开关(默认开启)
SHOW VARIABLES LIKE 'innodb_deadlock_detect';

-- 关闭死锁检测(不推荐)
SET GLOBAL innodb_deadlock_detect = OFF;

BEGIN;
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
-- 这里夹杂 RPC、库存检查、优惠计算、发送消息
-- 耗时可能长达数秒
UPDATE orders SET status = 'PAID' WHERE id = 1001;
COMMIT;

-- 方案1:缩小事务范围,只包含必要的数据库操作
BEGIN;
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
UPDATE orders SET status = 'PAID' WHERE id = 1001;
COMMIT;
-- 将 RPC、消息发送等移到事务外

-- 方案2:使用乐观锁,减少锁持有时间
SELECT version, status FROM orders WHERE id = 1001;
-- 执行业务逻辑(无锁)
BEGIN;
UPDATE orders 
SET status = 'PAID', version = version + 1 
WHERE id = 1001 AND version = 查询到的version;
COMMIT;

UPDATE orders
SET status = 'CANCELLED'
WHERE user_id = 1001 AND status = 'INIT';

-- 表结构
CREATE TABLE orders (
    id INT PRIMARY KEY,
    user_id INT,
    status VARCHAR(20),
    amount DECIMAL(10, 2),
    INDEX idx_user (user_id) -- 只有 user_id 的单列索引
);

-- 数据:用户 1001 有 1000 个订单

-- 事务 A
BEGIN;
UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 1001 AND status = 'INIT';
-- 需要扫描 user_id=1001 的所有订单
-- 在可重复读隔离级别下,可能锁定大量记录和间隙

-- 事务 B(同时执行)
BEGIN;
UPDATE orders SET status = 'PAID' 
WHERE user_id = 1001 AND id = 500;
-- 阻塞,因为被事务 A 的锁覆盖
COMMIT;

-- 方案1:创建联合索引
CREATE INDEX idx_user_status ON orders(user_id, status);

-- 方案2:分批更新
-- 每次只更新一小批,减少锁范围
UPDATE orders 
SET status = 'CANCELLED'
WHERE user_id = 1001 AND status = 'INIT'
LIMIT 100;

-- 方案3:使用主键范围更新
-- 先查询出需要更新的主键
SELECT id FROM orders 
WHERE user_id = 1001 AND status = 'INIT';
-- 然后按主键更新
UPDATE orders SET status = 'CANCELLED' WHERE id IN (1001, 1002, ...);

UPDATE stock
SET available = available - 1
WHERE sku_id = 9001 AND available > 0;

-- 创建多个库存桶
CREATE TABLE stock_bucket (
    id INT PRIMARY KEY,
    sku_id INT,
    bucket_no INT, -- 桶编号
    available INT,
    INDEX idx_sku_bucket (sku_id, bucket_no)
);

-- 初始化:sku_id=9001 的库存分散到 10 个桶
INSERT INTO stock_bucket VALUES
(1, 9001, 1, 100),
(2, 9001, 2, 100),
(3, 9001, 3, 100),
(4, 9001, 4, 100),
(5, 9001, 5, 100),
(6, 9001, 6, 100),
(7, 9001, 7, 100),
(8, 9001, 8, 100),
(9, 9001, 9, 100),
(10, 9001, 10, 100);

-- 扣库存时随机选择一个桶
UPDATE stock_bucket
SET available = available - 1
WHERE sku_id = 9001 AND bucket_no = FLOOR(1 + RAND() * 10)
AND available > 0;

-- 查询总库存时汇总
SELECT SUM(available) FROM stock_bucket WHERE sku_id = 9001;

// Redis 预扣库存
public boolean deductStock(Long skuId) {
    String key = "stock:" + skuId;
    Long remaining = redisTemplate.opsForValue().decrement(key);
    if (remaining < 0) {
        redisTemplate.opsForValue().increment(key);
        return false; // 库存不足
    }
    
    // 异步同步到数据库
    asyncUpdateDatabase(skuId);
    return true;
}

// 定期异步同步到数据库
@Scheduled(fixedRate = 1000)
public void syncStockToDatabase() {
    // 批量更新数据库
    batchUpdateStock();
}

-- 查询当前库存
SELECT id, available, version FROM stock WHERE sku_id = 9001;

-- 扣减库存(使用版本号)
UPDATE stock
SET available = available - 1, version = version + 1
WHERE sku_id = 9001 AND version = 查询到的version AND available > 0;

-- 如果影响行数为 0,说明并发冲突,业务重试

// 使用 Guava RateLimiter 限流
RateLimiter rateLimiter = RateLimiter.create(1000); // 每秒 1000 次

public boolean deductStock(Long skuId) {
    if (!rateLimiter.tryAcquire()) {
        return false; // 被限流
    }
    
    // 执行数据库扣减
    return doDeductStock(skuId);
}

-- 表结构
CREATE TABLE users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    username VARCHAR(50) UNIQUE,
    email VARCHAR(100)
);

-- 事务 A
BEGIN;
INSERT INTO users(username, email) VALUES('alice', 'alice@example.com');
-- 未提交

-- 事务 B(同时执行)
BEGIN;
INSERT INTO users(username, email) VALUES('alice', 'bob@example.com');
-- 阻塞,等待唯一索引锁
COMMIT;

-- 使用 INSERT IGNORE 忽略重复
INSERT IGNORE INTO users(username, email) VALUES('alice', 'bob@example.com');

-- 使用 ON DUPLICATE KEY UPDATE 更新
INSERT INTO users(username, email) VALUES('alice', 'bob@example.com')
ON DUPLICATE KEY UPDATE email = VALUES(email);

BEGIN;
UPDATE account SET balance = balance - 100 WHERE id = 1;
-- 执行其他业务逻辑
UPDATE account SET balance = balance + 100 WHERE id = 2;
COMMIT;

BEGIN;
UPDATE account SET balance = balance - 50 WHERE id = 2;
-- 执行其他业务逻辑
UPDATE account SET balance = balance + 50 WHERE id = 1;
COMMIT;

-- 统一加锁顺序:都按 id 从小到大加锁
BEGIN;
-- 先获取两个账户的锁,按 id 排序
SELECT * FROM account WHERE id IN (1, 2) ORDER BY id FOR UPDATE;

-- 然后执行转账
UPDATE account SET balance = balance - 100 WHERE id = 1;
UPDATE account SET balance = balance + 100 WHERE id = 2;
COMMIT;

BEGIN;
SELECT id FROM orders
WHERE status = 'INIT'
ORDER BY id
LIMIT 100
FOR UPDATE;

-- 处理这批订单
UPDATE orders SET status = 'PROCESSING' WHERE id IN (选中的id);
COMMIT;

BEGIN;
SELECT id FROM orders
WHERE status = 'INIT'
ORDER BY create_time -- 不同的排序字段
LIMIT 100
FOR UPDATE;

-- 处理这批订单
UPDATE orders SET status = 'PROCESSING' WHERE id IN (选中的id);
COMMIT;

-- 方案1:统一排序规则
-- 所有批量查询都使用相同的排序字段(如主键)
SELECT id FROM orders
WHERE status = 'INIT'
ORDER BY id
LIMIT 100
FOR UPDATE;

-- 方案2:使用分布式锁
-- 在应用层加分布式锁,保证同一时间只有一个批量任务执行

String lockKey = "batch_process_orders";
if (redisLock.tryLock(lockKey, 30, TimeUnit.SECONDS)) {
    try {
        // 执行批量处理
    } finally {
        redisLock.unlock(lockKey);
    }
}

-- 方案3:跳过锁策略
-- 使用 SKIP LOCKED 跳过已被锁定的记录(MySQL 8.0+)
SELECT id FROM orders
WHERE status = 'INIT'
ORDER BY id
LIMIT 100
FOR UPDATE SKIP LOCKED;

BEGIN;
INSERT INTO orders(id, user_id, status, amount) 
VALUES(1001, 100, 'INIT', 100.00);
COMMIT;

BEGIN;
UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 100 AND status = 'INIT';
COMMIT;

-- 方案1:缩小更新范围
-- 查询出具体的订单 id,然后按主键更新
SELECT id FROM orders 
WHERE user_id = 100 AND status = 'INIT';
UPDATE orders SET status = 'CANCELLED' WHERE id IN (具体id列表);

-- 方案2:调整隔离级别
-- 使用读已提交隔离级别,避免间隙锁
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;

-- 方案3:业务层面控制
-- 插入和批量更新错峰执行

-- 表结构
CREATE TABLE orders (
    id INT PRIMARY KEY,
    status VARCHAR(20),
    KEY idx_status (status)
);

CREATE TABLE order_items (
    id INT PRIMARY KEY,
    order_id INT,
    amount DECIMAL(10, 2),
    FOREIGN KEY (order_id) REFERENCES orders(id)
);

-- 事务 A
BEGIN;
UPDATE orders SET status = 'CANCELLED' WHERE id = 1001;
-- 需要检查子表 order_items 是否有关联记录
COMMIT;

-- 事务 B
BEGIN;
INSERT INTO order_items(id, order_id, amount) VALUES(1, 1001, 100.00);
-- 需要检查父表 orders 对应记录
COMMIT;

-- 方案1:避免使用外键约束
-- 在应用层维护数据一致性
-- 适合高并发场景

-- 方案2:统一操作顺序
-- 先操作父表,再操作子表
-- 所有涉及父子表的事务都遵循这个顺序

-- 方案3:使用存储过程封装操作
-- 在存储过程中使用事务和锁保证顺序
DELIMITER //
CREATE PROCEDURE update_order_with_items(
    IN p_order_id INT,
    IN p_status VARCHAR(20)
)
BEGIN
    DECLARE EXIT HANDLER FOR SQLEXCEPTION
    BEGIN
        ROLLBACK;
        RESIGNAL;
    END;
    
    START TRANSACTION;
    UPDATE orders SET status = p_status WHERE id = p_order_id;
    -- 其他操作
    COMMIT;
END //
DELIMITER ;

-- 表结构
CREATE TABLE logs (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id INT,
    action VARCHAR(50),
    KEY idx_user (user_id)
);

-- 数据:id = 1, 5, 10

-- 事务 A
BEGIN;
SELECT * FROM logs WHERE id = 3 FOR UPDATE;
-- id=3 不存在,可重复读下对间隙 (1, 5) 加间隙锁

-- 事务 B(同时执行)
BEGIN;
INSERT INTO logs(id, user_id, action) VALUES(3, 100, 'login');
-- 插入位置落在间隙 (1, 5) 内,插入意向锁被 A 的间隙锁阻塞,等待

-- 事务 A 继续
INSERT INTO logs(id, user_id, action) VALUES(3, 100, 'login');
-- A 也向间隙 (1, 5) 插入,该插入请求排在 B 的请求之后
-- B 在等 A 的间隙锁,A 在等排在队首的 B → 循环等待,死锁!
-- ERROR 1213 (40001): Deadlock found when trying to get lock

-- 方案1:降低隔离级别到 READ COMMITTED(无间隙锁,插入不会被间隙锁阻塞)
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;

-- 方案2:需要"锁住不存在的记录"时,先插入占位行,再对其加行锁
-- 把间隙锁问题变成普通行锁
INSERT INTO logs(id, user_id, action) VALUES(3, 100, 'init');
COMMIT;
BEGIN;
SELECT * FROM logs WHERE id = 3 FOR UPDATE; -- 行锁,不影响其他间隙的插入

-- 方案3:控制插入并发
-- 避免大量事务同时向同一索引范围插入,可按业务键分片路由

// Java 应用捕获死锁异常
try {
    orderService.processOrder(orderId);
} catch (DeadlockLoserDataAccessException e) {
    // 死锁异常,可以重试
    log.warn("Deadlock detected, retrying...", e);
    orderService.processOrder(orderId); // 重试
}

// Java 应用捕获锁等待超时异常(Spring 将 MySQL 1205 翻译为 CannotAcquireLockException)
try {
    orderService.processOrder(orderId);
} catch (CannotAcquireLockException e) {
    // 锁等待超时
    log.error("Lock wait timeout", e);
    throw new BusinessException("系统繁忙,请稍后重试");
}

SHOW ENGINE INNODB STATUS\G

LATEST DETECTED DEADLOCK
------------------------
2026-03-30 10:30:15 0x700012345000
*** (1) TRANSACTION:
TRANSACTION 12345, ACTIVE 2 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 100, OS thread handle 123456, query id 1000 localhost root updating
UPDATE account SET balance = balance - 100 WHERE id = 1
*** (1) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 58 page no 4 n bits 72 index PRIMARY of table `test`.`account` 
trx id 12345 lock_mode X locks rec but not gap waiting
Record lock, heap no 2 PHYSICAL RECORD: n_fields 3; compact format; info bits 0

*** (2) TRANSACTION:
TRANSACTION 12346, ACTIVE 1 sec starting index read
mysql tables in use 1, locked 1
3 lock struct(s), heap size 1136, 2 row lock(s), undo log entries 1
MySQL thread id 101, OS thread handle 123457, query id 1001 localhost root updating
UPDATE account SET balance = balance - 50 WHERE id = 2
*** (2) HOLDS THE LOCK(S):
RECORD LOCKS space id 58 page no 4 n bits 72 index PRIMARY of table `test`.`account` 
trx id 12346 lock_mode X locks rec but not gap
Record lock, heap no 2 PHYSICAL RECORD: n_fields 3; compact format; info bits 0
*** (2) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS space id 58 page no 5 n bits 72 index PRIMARY of table `test`.`account` 
trx id 12346 lock_mode X locks rec but not gap waiting
Record lock, heap no 3 PHYSICAL RECORD: n_fields 3; compact format; info bits 0

*** WE ROLL BACK TRANSACTION (1)

-- MySQL 8.0+:使用 sys.innodb_lock_waits 视图(底层来自 performance_schema.data_lock_waits)
SELECT waiting_pid, waiting_query, blocking_pid, blocking_query, wait_age_secs
FROM sys.innodb_lock_waits;

-- MySQL 5.7 及以下:使用 information_schema.innodb_lock_waits
SELECT 
    r.trx_id waiting_trx_id,
    r.trx_mysql_thread_id waiting_thread,
    r.trx_query waiting_query,
    b.trx_id blocking_trx_id,
    b.trx_mysql_thread_id blocking_thread,
    b.trx_query blocking_query
FROM information_schema.innodb_lock_waits w
INNER JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_id
INNER JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id;

SELECT 
    trx_id,
    trx_state,
    trx_started,
    TIMESTAMPDIFF(SECOND, trx_started, NOW()) AS duration_seconds,
    trx_mysql_thread_id,
    trx_query
FROM information_schema.innodb_trx
ORDER BY trx_started ASC;

-- MySQL 8.0+
SELECT 
    OBJECT_SCHEMA,
    OBJECT_NAME,
    LOCK_TYPE,
    LOCK_MODE,
    LOCK_STATUS,
    THREAD_ID,
    PROCESSLIST_ID
FROM performance_schema.data_locks
WHERE OBJECT_SCHEMA = 'your_database';

-- MySQL 5.7 及以下
SELECT * FROM information_schema.innodb_locks;

SHOW FULL PROCESSLIST;

-- 关注以下状态:
-- Waiting for table metadata lock: 等待元数据锁
-- Waiting for table level lock: 等待表锁
-- Waiting for lock: 等待行锁
-- Sending data: 可能是大查询
-- Statistics: 可能在计算执行计划

-- 开启慢查询日志
SET GLOBAL slow_query_log = 'ON';
SET GLOBAL long_query_time = 2; -- 超过 2 秒记录
SET GLOBAL log_queries_not_using_indexes = 'ON'; -- 记录未使用索引的查询

-- 查看慢查询日志
-- Linux: /var/log/mysql/slow.log
-- 或使用 mysqldumpslow 工具分析
mysqldumpslow -s t -t 10 /var/log/mysql/slow.log

// 在业务代码中记录事务执行时间
@Around("execution(* com.example.service.*.*(..))")
public Object logTransactionTime(ProceedingJoinPoint joinPoint) throws Throwable {
    long start = System.currentTimeMillis();
    try {
        return joinPoint.proceed();
    } finally {
        long duration = System.currentTimeMillis() - start;
        if (duration > 1000) { // 超过 1 秒记录
            log.warn("Slow transaction: {} took {} ms", 
                joinPoint.getSignature(), duration);
        }
    }
}

-- 检查表的索引
SHOW INDEX FROM orders;

-- 使用 EXPLAIN 分析 SQL
EXPLAIN UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 1001 AND status = 'INIT';

-- 检查索引使用情况
SELECT 
    TABLE_SCHEMA,
    TABLE_NAME,
    INDEX_NAME,
    CARDINALITY,
    SEQ_IN_INDEX
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = 'your_database'
ORDER BY TABLE_NAME, INDEX_NAME;

# 安装 Percona Toolkit
yum install percona-toolkit

# 使用 pt-deadlock-logger 持续监控死锁
pt-deadlock-logger h=localhost,u=root,p=password --daemonize --log /var/log/deadlock.log

# 使用 pt-query-digest 分析慢查询
pt-query-digest /var/log/mysql/slow.log

#!/bin/bash
# monitor_locks.sh - 监控锁等待情况

while true; do
    echo "=== $(date) ==="
    mysql -uroot -ppassword -e "
        SELECT 
            r.trx_id waiting_trx_id,
            r.trx_query waiting_query,
            b.trx_id blocking_trx_id,
            b.trx_query blocking_query,
            TIMESTAMPDIFF(SECOND, r.trx_started, NOW()) wait_seconds
        FROM information_schema.innodb_lock_waits w
        INNER JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_id
        INNER JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id
        WHERE TIMESTAMPDIFF(SECOND, r.trx_started, NOW()) > 5;
    "
    sleep 10
done

// × 错误示例:长事务
@Transactional
public void processOrder(Long orderId) {
    // 1. 查询订单
    Order order = orderRepository.findById(orderId);
    
    // 2. 检查库存(可能耗时)
    boolean stockAvailable = inventoryService.checkStock(order.getSkuId());
    if (!stockAvailable) {
        throw new BusinessException("库存不足");
    }
    
    // 3. 调用支付接口(可能耗时,可能失败)
    PaymentResult result = paymentService.pay(order);
    
    // 4. 发送消息通知(可能耗时)
    notificationService.sendNotification(order);
    
    // 5. 更新订单状态
    order.setStatus("PAID");
    orderRepository.save(order);
}

// √ 正确示例:缩短事务
public void processOrder(Long orderId) {
    // 事务外操作
    Order order = orderRepository.findById(orderId);
    
    boolean stockAvailable = inventoryService.checkStock(order.getSkuId());
    if (!stockAvailable) {
        throw new BusinessException("库存不足");
    }
    
    PaymentResult result = paymentService.pay(order);
    
    // 只在事务内执行必要的数据库操作
    updateOrderStatus(orderId, "PAID");
    
    // 异步发送通知
    notificationService.sendNotificationAsync(order);
}

@Transactional
public void updateOrderStatus(Long orderId, String status) {
    orderRepository.updateStatus(orderId, status);
}

// × 错误示例:大批量一次性处理
@Transactional
public void batchUpdateOrders(List<Long> orderIds) {
    List<Order> orders = orderRepository.findByIds(orderIds);
    for (Order order : orders) {
        order.setStatus("PROCESSING");
        // 其他复杂逻辑
    }
    orderRepository.saveAll(orders);
}

// √ 正确示例:分批处理
public void batchUpdateOrders(List<Long> orderIds) {
    // 分批处理,每批 100 条
    Lists.partition(orderIds, 100).forEach(batch -> {
        processBatch(batch);
    });
}

@Transactional
public void processBatch(List<Long> batch) {
    List<Order> orders = orderRepository.findByIds(batch);
    orders.forEach(order -> order.setStatus("PROCESSING"));
    orderRepository.saveAll(orders);
}

// × 错误示例:不同业务逻辑,加锁顺序不一致
// 业务 A:从账户 1 转账到账户 2
public void transferA(Long fromId, Long toId, BigDecimal amount) {
    accountRepository.deductBalance(fromId, amount);
    accountRepository.addBalance(toId, amount);
}

// 业务 B:从账户 2 转账到账户 1
public void transferB(Long fromId, Long toId, BigDecimal amount) {
    accountRepository.deductBalance(fromId, amount);
    accountRepository.addBalance(toId, amount);
}

// √ 正确示例:统一加锁顺序
public void transfer(Long fromId, Long toId, BigDecimal amount) {
    // 按 ID 排序,保证加锁顺序一致
    Long firstId = Math.min(fromId, toId);
    Long secondId = Math.max(fromId, toId);
    
    // 先锁定 ID 小的账户
    accountRepository.lockById(firstId);
    // 再锁定 ID 大的账户
    accountRepository.lockById(secondId);
    
    // 执行转账逻辑
    if (fromId < toId) {
        accountRepository.deductBalance(fromId, amount);
        accountRepository.addBalance(toId, amount);
    } else {
        accountRepository.addBalance(toId, amount);
        accountRepository.deductBalance(fromId, amount);
    }
}

// 使用 AOP 强制按顺序加锁
@Aspect
@Component
public class LockOrderAspect {
    
    @Around("@annotation(orderedLock)")
    public Object enforceLockOrder(ProceedingJoinPoint joinPoint, OrderedLock orderedLock) throws Throwable {
        // 获取方法参数中的 ID 列表
        Object[] args = joinPoint.getArgs();
        List<Long> ids = extractIds(args);
        
        // 排序后按顺序加锁
        Collections.sort(ids);
        for (Long id : ids) {
            lockManager.lock(id);
        }
        
        try {
            return joinPoint.proceed();
        } finally {
            // 按相反顺序释放锁
            Collections.reverse(ids);
            for (Long id : ids) {
                lockManager.unlock(id);
            }
        }
    }
}

// 使用注解标记需要按顺序加锁的方法
@OrderedLock
public void transfer(Long fromId, Long toId, BigDecimal amount) {
    // 业务逻辑
}

-- × 错误示例:更新条件无索引
CREATE TABLE orders (
    id INT PRIMARY KEY,
    user_id INT,
    status VARCHAR(20),
    create_time DATETIME
);

UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 1001 AND status = 'INIT';
-- 需要全表扫描,锁住大量记录

-- √ 正确示例:创建合适的索引
CREATE INDEX idx_user_status ON orders(user_id, status);

UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 1001 AND status = 'INIT';
-- 使用索引,只锁住满足条件的记录

-- 创建覆盖索引
CREATE INDEX idx_user_status_time ON orders(user_id, status, create_time);

-- 查询可以使用索引覆盖
SELECT id, create_time FROM orders 
WHERE user_id = 1001 AND status = 'INIT';
-- 不需要回表,性能更高

-- × 索引失效示例
-- 使用函数
UPDATE orders SET status = 'CANCELLED' 
WHERE DATE(create_time) = '2026-03-30';

-- 隐式类型转换
UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = '1001'; -- user_id 是 INT,字符串会导致索引失效

-- 使用 OR
UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 1001 OR status = 'INIT';

-- √ 正确示例
-- 使用范围查询
UPDATE orders SET status = 'CANCELLED' 
WHERE create_time >= '2026-03-30 00:00:00' 
AND create_time < '2026-03-31 00:00:00';

-- 类型匹配
UPDATE orders SET status = 'CANCELLED' 
WHERE user_id = 1001;

-- OR 两侧列都有索引时可能走 index_merge,否则拆成两条语句分别执行
UPDATE orders SET status = 'CANCELLED' WHERE user_id = 1001;
UPDATE orders SET status = 'CANCELLED' WHERE status = 'INIT' AND user_id != 1001;

-- 创建库存桶表
CREATE TABLE stock_bucket (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    sku_id BIGINT NOT NULL,
    bucket_no INT NOT NULL,
    available INT NOT NULL DEFAULT 0,
    total INT NOT NULL DEFAULT 0,
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_sku_bucket (sku_id, bucket_no),
    KEY idx_sku (sku_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 初始化库存(分 10 个桶)
DELIMITER //
CREATE PROCEDURE init_stock_buckets(
    IN p_sku_id BIGINT,
    IN p_total_stock INT,
    IN p_bucket_count INT
)
BEGIN
    DECLARE i INT DEFAULT 1;
    DECLARE per_bucket INT;
    SET per_bucket = CEIL(p_total_stock / p_bucket_count);
    
    WHILE i <= p_bucket_count DO
        INSERT INTO stock_bucket(sku_id, bucket_no, available, total)
        VALUES(p_sku_id, i, per_bucket, per_bucket);
        SET i = i + 1;
    END WHILE;
END //
DELIMITER ;

-- 初始化 sku_id=9001,总库存 1000,分 10 个桶
CALL init_stock_buckets(9001, 1000, 10);

-- 扣减库存(随机选择桶)
DELIMITER //
CREATE FUNCTION deduct_stock(
    p_sku_id BIGINT,
    p_quantity INT
) RETURNS BOOLEAN
NOT DETERMINISTIC
MODIFIES SQL DATA
BEGIN
    DECLARE v_bucket_no INT;
    DECLARE v_available INT;
    DECLARE v_affected INT DEFAULT 0;
    DECLARE retry_count INT DEFAULT 0;
    DECLARE max_retry INT DEFAULT 3;
    
    retry_loop: LOOP
        -- 随机选择一个桶
        SET v_bucket_no = FLOOR(1 + RAND() * 10);
        
        -- 检查库存
        SELECT available INTO v_available
        FROM stock_bucket
        WHERE sku_id = p_sku_id AND bucket_no = v_bucket_no;
        
        IF v_available >= p_quantity THEN
            -- 扣减库存
            UPDATE stock_bucket
            SET available = available - p_quantity
            WHERE sku_id = p_sku_id 
            AND bucket_no = v_bucket_no
            AND available >= p_quantity;
            
            SET v_affected = ROW_COUNT();
            
            IF v_affected > 0 THEN
                RETURN TRUE;
            END IF;
        END IF;
        
        SET retry_count = retry_count + 1;
        IF retry_count >= max_retry THEN
            RETURN FALSE;
        END IF;
    END LOOP;
    
    RETURN FALSE;
END //
DELIMITER ;

-- 查询总库存
SELECT sku_id, SUM(available) AS total_available
FROM stock_bucket
GROUP BY sku_id;

-- 库存同步(定期将各桶库存同步到主表)
DELIMITER //
CREATE PROCEDURE sync_stock_to_main()
BEGIN
    -- 同步到主库存表
    INSERT INTO stock(sku_id, available, total)
    SELECT sku_id, SUM(available), SUM(total)
    FROM stock_bucket
    GROUP BY sku_id
    ON DUPLICATE KEY UPDATE
    available = VALUES(available),
    total = VALUES(total);
END //
DELIMITER ;

// 使用 Spring Retry 实现自动重试
@Retryable(
    value = {DeadlockLoserDataAccessException.class, CannotAcquireLockException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 100, multiplier = 2)
)
@Transactional
public void processOrder(Long orderId) {
    Order order = orderRepository.findById(orderId);
    order.setStatus("PROCESSING");
    orderRepository.save(order);
}

// 重试失败后的兜底处理
@Recover
public void recover(DeadlockLoserDataAccessException e, Long orderId) {
    log.error("Deadlock retry failed for order: {}", orderId, e);
    // 发送告警
    alertService.sendAlert("Deadlock retry failed: " + orderId);
    // 记录到待处理队列
    pendingQueueService.add(orderId);
}

// 使用幂等键保证重试安全
@Transactional
public void processOrder(String idempotentKey, Long orderId) {
    // 检查幂等键是否已处理
    if (idempotentKeyRepository.existsByKey(idempotentKey)) {
        log.info("Order already processed: {}", idempotentKey);
        return;
    }
    
    // 执行业务逻辑
    Order order = orderRepository.findById(orderId);
    order.setStatus("PROCESSING");
    orderRepository.save(order);
    
    // 记录幂等键
    idempotentKeyRepository.save(new IdempotentKey(idempotentKey, orderId));
}

// 或者使用数据库唯一约束
@Transactional
public void processOrder(String idempotentKey, Long orderId) {
    try {
        // 插入幂等键(唯一约束)
        idempotentKeyRepository.insert(idempotentKey, orderId);
        
        // 执行业务逻辑
        Order order = orderRepository.findById(orderId);
        order.setStatus("PROCESSING");
        orderRepository.save(order);
    } catch (DuplicateKeyException e) {
        log.info("Order already processed: {}", idempotentKey);
        // 重复请求,直接返回
    }
}

// 完整的重试和兜底机制
@Service
@Slf4j
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private PendingQueueService pendingQueueService;
    
    @Autowired
    private AlertService alertService;
    
    private static final int MAX_RETRY = 3;
    
    public void processOrderWithRetry(Long orderId) {
        int retryCount = 0;
        
        while (retryCount < MAX_RETRY) {
            try {
                processOrder(orderId);
                return; // 成功,直接返回
            } catch (DeadlockLoserDataAccessException e) {
                retryCount++;
                log.warn("Deadlock detected for order: {}, retry count: {}", orderId, retryCount);
                
                if (retryCount >= MAX_RETRY) {
                    // 重试失败,记录到待处理队列
                    log.error("Max retry reached for order: {}", orderId);
                    pendingQueueService.add(orderId, "DEADLOCK_RETRY_FAILED");
                    alertService.sendAlert("Order process failed after retry: " + orderId);
                    return;
                }
                
                // 指数退避
                try {
                    Thread.sleep(100 * (1 << retryCount));
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    return;
                }
            } catch (Exception e) {
                // 其他异常,不重试
                log.error("Unexpected error for order: {}", orderId, e);
                throw new BusinessException("处理失败");
            }
        }
    }
    
    @Transactional
    public void processOrder(Long orderId) {
        Order order = orderRepository.findById(orderId);
        order.setStatus("PROCESSING");
        orderRepository.save(order);
    }
}

-- 情况1:唯一索引等值查询,只锁一行
UPDATE users SET name = 'Alice' WHERE id = 100;
-- 只锁 id=100 这一行

-- 情况2:范围查询,锁定多行和间隙
UPDATE users SET status = 'ACTIVE' WHERE id > 100 AND id < 200;
-- 锁定 id 在 (100, 200) 的所有行和间隙

-- 情况3:无索引,近似锁全表
UPDATE users SET status = 'ACTIVE' WHERE name = 'Alice';
-- name 列无索引,全表扫描,锁住所有行

-- 情况4:间隙锁,锁定间隙
SELECT * FROM users WHERE id > 100 FOR UPDATE;
-- 锁定 id > 100 的所有行和间隙

-- 查看当前事务持有的锁
SELECT * FROM performance_schema.data_locks 
WHERE THREAD_ID = CONNECTION_ID();

// √ 正确示例:有限重试 + 幂等保证
@Retryable(
    value = {DeadlockLoserDataAccessException.class},
    maxAttempts = 3, // 最多 3 次
    backoff = @Backoff(delay = 100, multiplier = 2) // 指数退避
)
@Transactional
public void processOrder(String idempotentKey, Long orderId) {
    // 幂等性检查
    if (isProcessed(idempotentKey)) {
        return;
    }
    
    // 业务逻辑
    doProcess(orderId);
    
    // 记录幂等键
    markAsProcessed(idempotentKey);
}

// × 错误示例:事务包含所有操作
@Transactional
public void processOrder(Long orderId) {
    // 1. 查询订单
    Order order = orderRepository.findById(orderId);
    
    // 2. 调用第三方接口(可能耗时数秒)
    PaymentResult result = paymentService.pay(order);
    
    // 3. 发送邮件通知
    emailService.sendNotification(order);
    
    // 4. 更新订单状态
    order.setStatus("PAID");
    orderRepository.save(order);
}

// √ 正确示例:缩小事务范围
public void processOrder(Long orderId) {
    // 事务外操作
    Order order = orderRepository.findById(orderId);
    PaymentResult result = paymentService.pay(order);
    emailService.sendNotification(order);
    
    // 只在事务内执行必要的数据库操作
    updateOrderStatus(orderId, "PAID");
}

@Transactional
public void updateOrderStatus(Long orderId, String status) {
    orderRepository.updateStatus(orderId, status);
}

// 使用消息队列实现最终一致性
public void processOrder(Long orderId) {
    // 1. 本地事务:创建订单,发送消息(事务消息)
    Order order = createOrder(orderId);
    
    // 2. 发送消息到 MQ
    messageQueueService.send("order.paid", order);
    
    // 3. 其他服务消费消息,执行相应逻辑
    // 库存服务:扣减库存
    // 通知服务:发送通知
}

// 库存服务消费消息
@RabbitListener(queues = "order.paid")
public void handleOrderPaid(Order order) {
    inventoryService.deductStock(order.getSkuId(), order.getQuantity());
}

-- 1. LOCK IN SHARE MODE(共享锁)
SELECT * FROM orders WHERE id = 1001 LOCK IN SHARE MODE;
-- 加共享锁,会阻塞写操作

-- 2. FOR UPDATE(排他锁)
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
-- 加排他锁,会阻塞所有操作

-- 3. 串行化隔离级别
SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
SELECT * FROM orders WHERE id = 1001;
-- 自动加共享锁

-- 快照读(不加锁)
SELECT * FROM orders WHERE id = 1001;
-- 读的是快照数据,不加锁

-- 当前读(加锁)
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
-- 读的是最新数据,加排他锁

UPDATE orders SET status = 'PAID' WHERE id = 1001;
-- 当前读,加排他锁

-- 事务 A
BEGIN;
SELECT * FROM orders WHERE id = 1001 LOCK IN SHARE MODE;
-- 加共享锁

-- 事务 B
BEGIN;
SELECT * FROM orders WHERE id = 1001 LOCK IN SHARE MODE;
-- 也加共享锁,兼容

-- 事务 A 继续
UPDATE orders SET status = 'PAID' WHERE id = 1001;
-- 需要排他锁,等待 B 释放共享锁

-- 事务 B 继续
UPDATE orders SET status = 'CANCELLED' WHERE id = 1001;
-- 也需要排他锁,等待 A 释放共享锁
-- 死锁!

-- √ 保持死锁检测开启
SHOW VARIABLES LIKE 'innodb_deadlock_detect';
-- Value: ON

-- √ 设置合理的锁等待超时时间
SET innodb_lock_wait_timeout = 10; -- 10 秒

-- √ 在应用层做好重试
-- 见 7.5 节重试机制

-- 通过主键更新:行锁
UPDATE users SET name = 'Alice' WHERE id = 1;

-- 无索引更新:全表扫描,锁住所有行,近似锁全表
UPDATE users SET name = 'Alice' WHERE name = 'Bob';

-- 表数据:id = 1, 5, 10
BEGIN;
SELECT * FROM users WHERE id > 3 AND id < 8 FOR UPDATE;
-- 锁定间隙 (1, 5) 和 (5, 10)

INSERT INTO users VALUES(2, 'test'); -- 阻塞
INSERT INTO users VALUES(8, 'test'); -- 阻塞

-- 查看死锁检测开关
SHOW VARIABLES LIKE 'innodb_deadlock_detect';

-- 关闭死锁检测(不推荐)
SET GLOBAL innodb_deadlock_detect = OFF;

SHOW ENGINE INNODB STATUS\G
-- 查看 LATEST DETECTED DEADLOCK 部分

-- 读已提交(RC):只锁记录
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT * FROM users WHERE id > 5 FOR UPDATE;
-- 只锁 id > 5 的已存在记录

-- 可重复读(RR):锁记录+间隙
SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT * FROM users WHERE id > 5 FOR UPDATE;
-- 锁定 (5, +∞) 的所有间隙和记录

-- 长事务:持有锁 5 秒
BEGIN;
SELECT * FROM orders WHERE id = 1001 FOR UPDATE;
-- 执行业务逻辑(5 秒)
UPDATE orders SET status = 'PAID' WHERE id = 1001;
COMMIT;

-- 影响:
-- 假设 QPS=100,5 秒内有 500 个请求
-- 这 500 个请求都会被阻塞

-- × 不同顺序:可能死锁
-- 事务 A
UPDATE account SET balance = balance - 100 WHERE id = 1;
UPDATE account SET balance = balance + 100 WHERE id = 2;

-- 事务 B
UPDATE account SET balance = balance - 50 WHERE id = 2;
UPDATE account SET balance = balance + 50 WHERE id = 1;

-- √ 统一顺序:不会死锁
-- 都按 id 从小到大
SELECT * FROM account WHERE id IN (1, 2) ORDER BY id FOR UPDATE;
-- 然后执行转账

*** (1) TRANSACTION:
TRANSACTION 12345, ACTIVE 2 sec
UPDATE account SET balance = balance - 100 WHERE id = 1
*** (1) WAITING FOR THIS LOCK TO BE GRANTED:
RECORD LOCKS ... lock_mode X ... waiting

*** (2) TRANSACTION:
TRANSACTION 12346, ACTIVE 1 sec
UPDATE account SET balance = balance - 50 WHERE id = 2
*** (2) HOLDS THE LOCK(S):
RECORD LOCKS ... lock_mode X

-- 解读:
-- 事务 1 执行 UPDATE id=1,等待锁
-- 事务 2 执行 UPDATE id=2,持有锁
-- 需要看后续信息确认循环等待

1. 用户请求 → 限流检查
2. Redis 扣库存 → 成功/失败
3. 成功 → 创建订单(异步)
4. 定时任务 → 同步 Redis 到数据库
5. 数据库分桶 → 提升并发能力