锁冲突与死锁案例

-- 事务 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),
...
(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;
INSERT INTO logs(user_id, action) VALUES(100, 'login');
-- 在间隙 (1, 5) 插入,获取 id=2

-- 事务 B
BEGIN;
INSERT INTO logs(user_id, action) VALUES(200, 'logout');
-- 也在间隙 (1, 5) 插入,获取 id=3

-- 事务 C
BEGIN;
SELECT * FROM logs WHERE id > 1 AND id < 5 FOR UPDATE;
-- 尝试锁定间隙 (1, 5)

-- 方案1:使用顺序自增
-- innodb_autoinc_lock_mode = 0 (传统模式)
-- 所有插入都串行,避免间隙锁冲突(性能低)

-- 方案2:使用 UUID 或雪花算法生成主键
-- 避免自增主键的间隙问题
CREATE TABLE logs (
    id BIGINT PRIMARY KEY, -- 使用雪花算法生成
    user_id INT,
    action VARCHAR(50)
);

-- 方案3:批量插入使用单个事务
-- 减少并发插入的事务数量
BEGIN;
INSERT INTO logs(user_id, action) VALUES 
(100, 'login'), (200, 'logout'), (300, 'click');
COMMIT;

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

// Java 应用捕获锁等待超时异常
try {
    orderService.processOrder(orderId);
} catch (LockTimeoutException 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+ 使用 performance_schema
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;

-- MySQL 5.7 及以下
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;

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

-- 创建库存桶表
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
DETERMINISTIC
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, LockTimeoutException.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. 数据库分桶 → 提升并发能力