悲观锁引发死锁的常见原因与有效防范策略
嗨，朋友！今天咱们来聊聊编程世界里那个让人头疼的问题——悲观锁引发的死锁。作为一个摸爬滚打多年的开发者，我见过太多人在这上面栽跟头，今天就把我的经验都掏给你。
一、先搞清楚：什么是悲观锁？
悲观锁这东西，说简单点就是——”我总觉得别人会来抢我的数据，所以我先锁上再说”。
想象一下你去图书馆占座，你放一本书在位置上，然后去上厕所。在你回来之前，别人是不能坐这个位置的。这就是悲观锁的思想：先加锁，再操作。
在代码世界里，悲观锁通常表现为数据库的 SELECT ... FOR UPDATE，或者编程语言中的 synchronized、mutex 等机制。
-- MySQL 中的悲观锁示例
BEGIN;
SELECT * FROM account WHERE id = 1 FOR UPDATE; -- 加上悲观锁
UPDATE account SET balance = balance - 100 WHERE id = 1;
COMMIT;
这条SQL在执行SELECT ... FOR UPDATE时，会给这一行加排他锁（X锁），其他事务想要修改这一行，就必须等待当前事务提交或回滚。
二、死锁是怎么发生的？
死锁这个词听起来很高大上，其实它的本质特别简单：两个（或更多）事务互相等待，谁也不肯先放手，结果大家都卡死了。
让我用一个经典的例子来说明：
场景：转账操作
假设有两个账户A和B，现在有两个事务同时在执行转账：
- 事务1：把A账户的钱转到B账户
- 事务2：把B账户的钱转到A账户
如果没有死锁，它们应该这样执行：
事务1: 锁住A → 锁住B → 完成转账 → 提交
事务2: 等事务1完成后 → 锁住B → 锁住A → 完成转账 → 提交
但现实往往不是这样……
三、悲观锁引发死锁的常见原因
原因1：锁获取顺序不一致
这是最常见、也是最容易被忽视的原因。
来看这段代码：
// 事务1：从A账户转账到B账户
@Transactional
public void transferAB(Long accountIdA, Long accountIdB, BigDecimal amount) {
// 先锁A账户
Account accountA = accountRepository.findByIdForUpdate(accountIdA);
// 再锁B账户
Account accountB = accountRepository.findByIdForUpdate(accountIdB);
// 执行转账逻辑
accountA.setBalance(accountA.getBalance().subtract(amount));
accountB.setBalance(accountB.getBalance().add(amount));
accountRepository.update(accountA);
accountRepository.update(accountB);
}
// 事务2：从B账户转账到A账户
@Transactional
public void transferBA(Long accountIdB, Long accountIdA, BigDecimal amount) {
// 先锁B账户
Account accountB = accountRepository.findByIdForUpdate(accountIdB);
// 再锁A账户
Account accountA = accountRepository.findByIdForUpdate(accountIdA);
// 执行转账逻辑
accountB.setBalance(accountB.getBalance().subtract(amount));
accountA.setBalance(accountA.getBalance().add(amount));
accountRepository.update(accountB);
accountRepository.update(accountA);
}
现在想象这两个事务同时执行：
时间点 事务1 事务2
─────────────────────────────────────────────────
T1 锁住账户A 锁住账户B
T2 请求锁账户B（被阻塞） 请求锁账户A（被阻塞）
T3 ...等待... ...等待...
T4 死锁！ 死锁！
事务1拿着A的锁等B的锁，事务2拿着B的锁等A的锁。两个事务互相等待，谁也不能继续，数据库检测到这个循环等待后，会选择牺牲其中一个事务（通常是成本较小的那个），抛出死锁异常。
这就是典型的”锁顺序不一致”导致的死锁。
原因2：锁持有时间过长
悲观锁一旦获取，就需要一直持有到事务提交。如果事务执行时间太长，或者在持有锁期间做了大量不必要的操作，就会增加死锁的概率。
// 错误示例：锁持有时间过长
@Transactional
public void processOrder(Order order) {
// 获取悲观锁
Order orderDetail = orderRepository.findByIdForUpdate(order.getId());
// 这里做了一些耗时的操作，比如调用外部API
// 在持有锁的情况下调用外部服务，锁会一直持有
String result = externalService.call(orderDetail);
// 又做了一些数据库查询
List<Item> items = itemRepository.findAll();
// 最后才更新数据
orderDetail.setStatus("PROCESSED");
orderRepository.update(orderDetail);
}
这种代码的问题在于：锁在外部调用和无关查询期间一直被持有，其他需要访问同一行数据的事务只能干等。等待时间越长，死锁的可能性就越大。
原因3：锁粒度太粗
有时候，开发者为了图省事，直接锁了整个表或者整个查询结果集，而不是锁具体的行。
-- 粗粒度的锁，锁住了整个表
SELECT * FROM orders WHERE status = 'PENDING' FOR UPDATE;
这条语句会锁住所有满足条件的行，甚至在一些数据库实现中会锁表。如果有多个事务同时执行类似的查询，很容易形成死锁。
想象一下：
- 事务1锁住了状态为”PENDING”的所有订单（假设是订单1、2、3）
- 事务2也锁住了状态为”PENDING”的所有订单
- 两个事务互相覆盖，冲突概率大增
原因4：嵌套锁和多层依赖
当系统比较复杂时，一个事务可能会锁住多个资源，而这些资源又被其他事务以不同顺序锁住。
事务1: 锁资源A → 锁资源B → 锁资源C
事务2: 锁资源B → 锁资源C → 锁资源A
事务3: 锁资源C → 锁资源A → 锁资源B
这种三层嵌套锁的情况，死锁的概率会呈指数级增长。任何一个时刻，只要三个事务的执行顺序稍有偏差，就可能形成循环等待。
原因5：数据库隔离级别设置不当
数据库的隔离级别直接影响锁的行为。在较高的隔离级别（如 Serializable）下，悲观锁的行为会更加严格，死锁的概率也会增加。
-- 设置事务隔离级别为Serializable
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
BEGIN;
SELECT * FROM account WHERE id = 1 FOR UPDATE;
-- 在这个隔离级别下，甚至读操作也可能加锁
COMMIT;
Serializable是最高隔离级别，它通过严格的锁机制来保证事务的串行执行，但这也意味着死锁的可能性大大增加。
四、如何有效防范悲观锁引发的死锁？
好了，原因分析完了，现在来看看怎么解决。作为过来人，我给你整理了几条实战中验证过的策略。
策略1：统一锁的获取顺序
这是最简单、最有效的方法。无论什么场景，都按照固定的顺序获取锁，通常是按照资源的ID从小到大排序。
/**
* 修复后的转账方法：统一锁获取顺序
*/
@Transactional
public void transfer(Long fromAccountId, Long toAccountId, BigDecimal amount) {
// 关键：始终按ID顺序获取锁，避免死锁
Long firstId = Math.min(fromAccountId, toAccountId);
Long secondId = Math.max(fromAccountId, toAccountId);
// 先锁ID小的账户
Account firstAccount = accountRepository.findByIdForUpdate(firstId);
// 再锁ID大的账户
Account secondAccount = accountRepository.findByIdForUpdate(secondId);
// 找到对应的转出和转入账户
Account fromAccount = fromAccountId.equals(firstId) ? firstAccount : secondAccount;
Account toAccount = fromAccountId.equals(firstId) ? secondAccount : firstAccount;
// 执行转账
if (fromAccount.getBalance().compareTo(amount) < 0) {
throw new InsufficientBalanceException("余额不足");
}
fromAccount.setBalance(fromAccount.getBalance().subtract(amount));
toAccount.setBalance(toAccount.getBalance().add(amount));
accountRepository.update(fromAccount);
accountRepository.update(toAccount);
}
这样一来，无论哪个事务执行转账，锁的顺序都是固定的（先锁小ID，再锁大ID），从根本上消除了循环等待的可能性。
策略2：缩短锁持有时间
尽量让锁的持有时间最短，把非必要的操作移到锁外面。
/**
* 优化后的订单处理方法：缩短锁持有时间
*/
@Transactional
public void processOrder(Long orderId) {
// 先获取需要处理的数据
Order order = orderRepository.findByIdForUpdate(orderId);
// 立即释放锁（在同一个事务内，但后续操作不再需要锁）
// 注意：这里利用的是乐观锁的思想，或者将后续操作移出事务
// 在事务外执行耗时操作
String result = externalService.call(order);
// 重新开启一个短事务来更新数据
updateOrderStatus(orderId, result);
}
/**
* 短事务：只执行必要的数据库操作
*/
@Transactional
public void updateOrderStatus(Long orderId, String result) {
Order order = orderRepository.findByIdForUpdate(orderId);
order.setStatus("PROCESSED");
order.setResult(result);
orderRepository.update(order);
}
如果业务场景不允许拆分事务，至少也要确保在持有锁期间只做必要的数据库操作：
@Transactional
public void processOrder(Order order) {
// 获取悲观锁
Order orderDetail = orderRepository.findByIdForUpdate(order.getId());
// 只在锁持有期间执行必要的数据库操作
orderDetail.setStatus("PROCESSING");
orderDetail.setProcessedAt(LocalDateTime.now());
orderRepository.update(orderDetail);
// 锁已释放，再执行外部调用
String result = externalService.call(orderDetail);
// 如果需要更新结果，再开启一个新事务
updateOrderResult(orderDetail.getId(), result);
}
策略3：使用更细粒度的锁
尽量锁住具体的行，而不是整个表或大面积的数据范围。
-- 差：锁住所有PENDING状态的订单
SELECT * FROM orders WHERE status = 'PENDING' FOR UPDATE;
-- 好：只锁住具体的订单
SELECT * FROM orders WHERE id = 12345 FOR UPDATE;
如果确实需要批量处理，可以考虑分批获取锁：
/**
* 分批处理订单，避免一次锁住太多数据
*/
@Transactional
public void processPendingOrders() {
// 每次只获取一小批订单进行处理
List<Order> batch = orderRepository.findPendingOrdersForUpdate(10);
for (Order order : batch) {
try {
processSingleOrder(order);
} catch (Exception e) {
// 处理单个订单失败，记录日志但不影响其他订单
logger.error("处理订单{}失败", order.getId(), e);
}
}
}
/**
* 处理单个订单，锁粒度最细
*/
@Transactional
public void processSingleOrder(Order order) {
// 只锁当前订单
Order lockedOrder = orderRepository.findByIdForUpdate(order.getId());
// 执行处理逻辑
lockedOrder.setStatus("PROCESSED");
orderRepository.update(lockedOrder);
}
策略4：设置合理的超时时间
即使做了所有预防工作，死锁仍可能因为某些边界情况发生。这时候，合理的超时机制就显得尤为重要。
/**
* 带超时机制的转账操作
*/
public void transferWithTimeout(Long fromAccountId, Long toAccountId,
BigDecimal amount, long timeoutMillis) {
// 尝试获取锁，设置超时时间
Lock lock = getLock(fromAccountId);
boolean acquired = false;
try {
acquired = lock.tryLock(timeoutMillis, TimeUnit.MILLISECONDS);
if (!acquired) {
// 超时无法获取锁，返回失败信息
throw new LockTimeoutException("获取账户" + fromAccountId + "的锁超时");
}
// 获取第二个账户的锁
Lock lock2 = getLock(toAccountId);
boolean acquired2 = lock2.tryLock(timeoutMillis, TimeUnit.MILLISECONDS);
if (!acquired2) {
lock.unlock();
throw new LockTimeoutException("获取账户" + toAccountId + "的锁超时");
}
try {
// 执行转账逻辑
executeTransfer(fromAccountId, toAccountId, amount);
} finally {
lock2.unlock();
}
} finally {
if (acquired) {
lock.unlock();
}
}
}
在数据库层面，也可以设置锁等待超时：
-- MySQL：设置锁等待超时时间（单位：秒）
SET innodb_lock_wait_timeout = 50;
-- 或者在事务级别设置
SET SESSION innodb_lock_wait_timeout = 50;
当某个事务等待锁的时间超过设定值时，数据库会自动中止该事务，避免长时间占用资源。
策略5：使用乐观锁替代悲观锁
在很多场景下，乐观锁是比悲观锁更好的选择。乐观锁的思想是：先操作，出问题了再解决。
/**
* 使用乐观锁的转账实现
*/
@OptimisticLocking
@Transactional
public void optimisticTransfer(Long fromAccountId, Long toAccountId,
BigDecimal amount) {
// 不加锁，直接读取数据
Account fromAccount = accountRepository.findById(fromAccountId);
Account toAccount = accountRepository.findById(toAccountId);
// 执行转账逻辑
if (fromAccount.getBalance().compareTo(amount) < 0) {
throw new InsufficientBalanceException("余额不足");
}
fromAccount.setBalance(fromAccount.getBalance().subtract(amount));
toAccount.setBalance(toAccount.getBalance().add(amount));
// 更新时，乐观锁会检查版本号
// 如果版本号不匹配（说明有并发修改），会抛出异常
accountRepository.update(fromAccount);
accountRepository.update(toAccount);
}
对应的数据库表结构需要加上版本号：
CREATE TABLE account (
id BIGINT PRIMARY KEY,
balance DECIMAL(19, 2),
version INT DEFAULT 0 -- 乐观锁版本号
);
更新语句会自动带上版本号条件：
UPDATE account
SET balance = balance - 100, version = version + 1
WHERE id = 1 AND version = 5;
如果返回影响行数为0，说明有其他事务已经修改了这条记录，此时可以重试。
/**
* 带重试机制的乐观锁操作
*/
public void transferWithRetry(Long fromAccountId, Long toAccountId,
BigDecimal amount, int maxRetries) {
for (int i = 0; i < maxRetries; i++) {
try {
optimisticTransfer(fromAccountId, toAccountId, amount);
return; // 成功则返回
} catch (OptimisticLockException e) {
// 乐观锁冲突，重试
if (i == maxRetries - 1) {
throw e; // 超过最大重试次数，抛出异常
}
logger.warn("乐观锁冲突，第{}次重试", i + 1);
}
}
}
乐观锁的优势在于：不加锁，所以不会死锁。它的代价是可能会有并发冲突，需要通过重试机制来处理。在大多数业务场景中，这个代价是可以接受的。
策略6：使用数据库死锁检测机制
现代数据库（如MySQL的InnoDB引擎）都内置了死锁检测机制。当检测到死锁时，数据库会自动选择一个”牺牲品”事务回滚，让其他事务继续执行。
-- 查看InnoDB死锁信息
SHOW ENGINE INNODB STATUS;
-- 在结果中查找"LATEST DETECTED DEADLOCK"部分
你也可以调整死锁检测的相关参数：
-- 查看当前死锁检测相关配置
SHOW VARIABLES LIKE 'innodb%';
-- innodb_deadlock_detect: 是否启用死锁检测（默认ON）
-- innodb_lock_wait_timeout: 锁等待超时时间（默认50秒）
-- innodb_commit_concurrency: 并发提交事务数
虽然数据库自带了死锁检测，但依赖检测机制并不是一个好的做法。预防永远比补救好，与其让数据库来检测和解决死锁，不如从一开始就避免死锁的发生。
五、实战案例分析
让我给你讲一个真实的案例，看看这些策略是如何应用的。
案例背景
我们有一个电商系统，核心功能是订单创建和库存扣减。系统上线后，在促销活动期间频繁出现死锁错误，导致部分订单创建失败。
问题分析
通过查看数据库日志，我们发现死锁的模式非常一致：
LATEST DETECTED DEADLOCK
------------------------
2024-03-15 10:23:45 7E8A9B00
*** (1) TRANSACTION:
TRANSACTION 12345678, ACTIVE 0 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 890123, OS thread handle 140123456789, query id 987654321
UPDATE orders SET status='PAID' WHERE id=12345
*** (2) TRANSACTION:
TRANSACTION 12345679, ACTIVE 0 sec starting index read
mysql tables in use 1, locked 1
2 lock struct(s), heap size 1136, 1 row lock(s)
MySQL thread id 890124, OS thread handle 140123456789, query id 987654322
UPDATE inventory SET stock=stock-1 WHERE product_id=67890
*** WE ROLL BACK TRANSACTION (2)
问题分析：
- 两个事务分别操作不同的表，但存在间接依赖关系
- 事务1先锁订单表，再锁库存表
- 事务2先锁库存表，再锁订单表
- 锁顺序不一致导致死锁
解决方案
我们采取了以下措施：
措施一：统一锁获取顺序
/**
* 创建订单并扣减库存
*/
@Transactional
public Order createOrder(OrderRequest request) {
// 统一按照表名的字母顺序获取锁：inventory -> orders
// 先锁库存
Inventory inventory = inventoryRepository
.findByNameForUpdate(request.getProductName());
// 再锁订单
Order order = new Order();
order.setProductName(request.getProductName());
order.setStatus("PENDING");
orderRepository.save(order);
// 扣减库存
inventory.setStock(inventory.getStock() - request.getQuantity());
inventoryRepository.update(inventory);
// 更新订单状态
order.setStatus("PAID");
orderRepository.update(order);
return order;
}
措施二：使用乐观锁处理库存
/**
* 使用乐观锁扣减库存
*/
public void reduceStockOptimistic(String productName, int quantity) {
for (int i = 0; i < MAX_RETRIES; i++) {
Inventory inventory = inventoryRepository
.findByName(productName);
if (inventory.getStock() < quantity) {
throw new InsufficientStockException("库存不足");
}
inventory.setStock(inventory.getStock() - quantity);
int updated = inventoryRepository.updateWithVersion(
inventory.getId(),
inventory.getStock(),
inventory.getVersion()
);
if (updated > 0) {
return; // 成功
}
// 乐观锁冲突，重试
}
}
措施三：设置合理的超时时间
# application.yml 配置
spring:
datasource:
hikari:
connection-timeout: 30000 # 连接超时30秒
idle-timeout: 600000 # 空闲连接超时10分钟
max-lifetime: 1800000 # 连接最大存活时间30分钟
# 数据库层面设置
# innodb_lock_wait_timeout = 10 # 10秒后超时
措施四：添加死锁监控和告警
/**
* 死锁监控服务
*/
@Component
public class DeadlockMonitor {
@Scheduled(fixedRate = 60000) // 每分钟检查一次
public void checkDeadlocks() {
try {
// 查询死锁信息
String status = databaseService
.executeQuery("SHOW ENGINE INNODB STATUS");
// 解析死锁信息
if (status.contains("LATEST DETECTED DEADLOCK")) {
// 记录详细日志
logger.error("检测到死锁: {}", status);
// 发送告警
alertService.sendAlert("系统检测到死锁，请及时处理");
}
} catch (Exception e) {
logger.error("死锁检查失败", e);
}
}
}
实施效果
经过以上优化，死锁问题得到了根本解决：
- 生产环境死锁次数从每天数十次降为零
- 订单创建成功率从95%提升到99.9%
- 系统在高并发场景下的稳定性显著提升
六、总结与建议
聊了这么多，我来给你总结一下核心要点：
死锁产生的四个必要条件
记住这四个条件，你就能理解为什么会产生死锁，以及如何破坏它们：
- 互斥条件：资源不能共享，每次只能被一个事务占用
- 请求与保持条件：事务持有资源的同时，请求其他资源
- 不剥夺条件：事务获得的资源在使用完之前不能被剥夺
- 循环等待条件：存在一个事务循环等待链
只要破坏其中任何一个条件，就能避免死锁。
我的建议
优先使用乐观锁：在大多数业务场景下，乐观锁是更好的选择。它不会死锁，性能也更好。只有在中高冲突率且对数据一致性要求极高的场景下，才考虑使用悲观锁。
如果必须用悲观锁，就遵循以下原则：
- 统一锁的获取顺序
- 尽量缩短锁持有时间
- 使用细粒度的锁
- 设置合理的超时时间
做好监控和告警：即使做了所有预防工作，死锁仍可能因为边界情况发生。建立监控机制，及时发现和处理问题。
测试！测试！测试！：在上线前，一定要进行充分的压力测试和并发测试，模拟高并发场景，验证系统是否会死锁。
/**
* 并发测试示例
*/
@Test
public void testConcurrentTransfer() throws InterruptedException {
int threadCount = 100;
CountDownLatch latch = new CountDownLatch(threadCount);
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger deadlockCount = new AtomicInteger(0);
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < threadCount; i++) {
final int fromId = i % 10 + 1;
final int toId = (i + 1) % 10 + 1;
Thread thread = new Thread(() -> {
try {
transfer(fromId, toId, BigDecimal.ONE);
successCount.incrementAndGet();
} catch (DeadlockException e) {
deadlockCount.incrementAndGet();
} finally {
latch.countDown();
}
});
threads.add(thread);
thread.start();
}
latch.await();
System.out.println("成功: " + successCount.get());
System.out.println("死锁: " + deadlockCount.get());
// 断言：不应该有死锁
assertEquals(0, deadlockCount.get());
}
好了，今天就聊到这里。悲观锁和死锁这个话题说大不大，说小不小，但只要理解了它的本质，掌握了正确的使用方法，就能在代码世界里游刃有余。记住：好的代码不仅要能运行，还要能在高并发下稳定运行。
如果你在实际开发中遇到了死锁问题，欢迎随时来找我交流。编程这条路，咱们一起走！