老王因锁顺序混乱死锁赔了50万用统一加锁顺序与超时重试彻底解决悲观锁死锁难题

那天下午，老王坐在会议室里，脸色比他的代码还要灰暗。对面坐着的是公司CTO和法务，桌上摊着一份50万的赔偿协议。事情起因很简单——一个看似不起眼的高并发下单接口，因为两个线程抢锁的顺序不一致，导致了死锁。数据库连接池里的线程全部卡死，订单系统宕机了整整40分钟。

这50万，赔的是客户的损失，也是老王职业生涯的一记警钟。

死锁：程序员最讨厌的”幽灵”

什么叫死锁？简单来说，就是两个或更多的线程互相持有对方需要的锁，然后谁也不让谁，大家就这么僵持着，直到天荒地老。

想象一下这个场景：

老王负责维护一个库存系统。有两个操作：

A操作：从仓库1扣库存，再往仓库2补库存
B操作：从仓库2扣库存，再往仓库1补库存

如果A和B同时执行，可能会发生这样的事：

时间线 ->

线程A: 拿到仓库1的锁 --> 准备拿仓库2的锁 (等待中)
线程B: 拿到仓库2的锁 --> 准备拿仓库1的锁 (等待中)

结果：线程A等线程B释放仓库2的锁
      线程B等线程A释放仓库1的锁
      双方永远等下去 = 死锁

这就是经典的”循环等待”。死锁发生的四个必要条件，任何一个破坏了就能避免死锁：

互斥条件：锁一次只能被一个线程持有
请求与保持条件：线程持有锁的同时还在请求其他锁
不剥夺条件：锁不能被强制剥夺
循环等待条件：存在一个线程等待的循环链

老王当年的问题，就坏在第4条——循环等待。

老王当年的”血泪代码”

让我们看看老王当年到底写了什么：

// 仓库1和仓库2的锁
private final ReentrantLock warehouse1Lock = new ReentrantLock();
private final ReentrantLock warehouse2Lock = new ReentrantLock();

// 操作A：从1转移到2
public void transferA() {
    warehouse1Lock.lock();
    try {
        // 做一些业务逻辑...
        
        warehouse2Lock.lock();  // 这里可能死锁！
        try {
            // 更新仓库2
        } finally {
            warehouse2Lock.unlock();
        }
    } finally {
        warehouse1Lock.unlock();
    }
}

// 操作B：从2转移到1
public void transferB() {
    warehouse2Lock.lock();
    try {
        // 做一些业务逻辑...
        
        warehouse1Lock.lock();  // 这里可能死锁！
        try {
            // 更新仓库1
        } finally {
            warehouse1Lock.unlock();
        }
    } finally {
        warehouse2Lock.unlock();
    }
}

看到问题了吗？

transferA 先拿仓库1的锁，再拿仓库2的锁
transferB 先拿仓库2的锁，再拿仓库1的锁

当两个方法同时执行时，完美的死锁就形成了。

更糟糕的是，老王还加了一个”智能优化”——他给锁加了超时：

// 自以为是的优化
warehouse2Lock.tryLock(100, TimeUnit.MILLISECONDS);

但他忘了，tryLock返回false时，他并没有回滚已经持有的warehouse1Lock！这导致锁一直被持有，系统越来越卡。

解决方案一：统一加锁顺序

最经典、最有效的方案，就是规定一个全局的加锁顺序。

不管哪个线程，不管什么场景，必须按照固定的顺序加锁。比如，我们规定：永远先锁ID小的仓库，再锁ID大的仓库。

import java.util.concurrent.locks.ReentrantLock;

public class DeadlockSafeInventory {
    
    private final ReentrantLock warehouse1Lock = new ReentrantLock();
    private final ReentrantLock warehouse2Lock = new ReentrantLock();
    
    // 定义锁的顺序，比如按ID排序
    private static final ReentrantLock[] LOCKS = {
        warehouse1Lock,  // ID=1
        warehouse2Lock   // ID=2
    };
    
    /**
     * 统一的加锁方法：始终按ID顺序加锁
     */
    private void lockOrders(ReentrantLock first, ReentrantLock second) {
        // 比较锁的标识（可以通过ID或哈希值）
        if (System.identityHashCode(first) < System.identityHashCode(second)) {
            first.lock();
            try {
                second.lock();
            } catch (Exception e) {
                first.unlock();
                throw e;
            }
        } else {
            second.lock();
            try {
                first.lock();
            } catch (Exception e) {
                second.unlock();
                throw e;
            }
        }
    }
    
    /**
     * 解锁也要按相反顺序
     */
    private void unlockOrders(ReentrantLock first, ReentrantLock second) {
        second.unlock();
        first.unlock();
    }
    
    // 操作A：从1转移到2
    public void transferA(int amount) {
        lockOrders(warehouse1Lock, warehouse2Lock);
        try {
            // 业务逻辑：扣仓库1，加仓库2
            System.out.println(Thread.currentThread().getName() + " 执行转移A");
        } finally {
            unlockOrders(warehouse1Lock, warehouse2Lock);
        }
    }
    
    // 操作B：从2转移到1
    public void transferB(int amount) {
        lockOrders(warehouse1Lock, warehouse2Lock);  // 注意：也是这个顺序！
        try {
            // 业务逻辑：扣仓库2，加仓库1
            System.out.println(Thread.currentThread().getName() + " 执行转移B");
        } finally {
            unlockOrders(warehouse1Lock, warehouse2Lock);
        }
    }
}

现在，不管你是transferA还是transferB，加锁顺序都是一致的：先锁warehouse1，再锁warehouse2。循环等待的条件被打破了，死锁不可能发生。

为什么统一顺序能解决问题？

画个时间线你就明白了：

原本（会死锁）：
线程A: [锁1] --> [等锁2]
线程B: [锁2] --> [等锁1]
          ↑_________↓
          互相等待，死锁！

统一顺序后：
线程A: [锁1] --> [锁2] --> 完成 --> 解锁
线程B: [锁1] --> [锁2] --> 完成 --> 解锁

线程B必须等线程A释放锁1，才能继续。不会再有”你等我、我等你”的僵局。

解决方案二：超时重试机制

光有顺序还不够，现实世界很复杂。有时候锁的顺序不好确定，有时候业务逻辑动态变化。这时候，超时重试就是最后一道保险。

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReentrantLock;

public class RobustInventorySystem {
    
    private final ReentrantLock warehouse1Lock = new ReentrantLock();
    private final ReentrantLock warehouse2Lock = new ReentrantLock();
    
    private static final int MAX_RETRY = 3;
    private static final long LOCK_TIMEOUT_MS = 100;
    
    /**
     * 带超时的加锁，失败则回滚已持有的锁并重试
     */
    public <T> T executeWithLocks(RunnableWithResult<T> action) {
        for (int attempt = 0; attempt < MAX_RETRY; attempt++) {
            boolean locked1 = false;
            boolean locked2 = false;
            
            try {
                // 按统一顺序加锁
                warehouse1Lock.lockInterruptibly();
                locked1 = true;
                
                // 尝试获取第二个锁，带超时
                locked2 = warehouse2Lock.tryLock(LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
                
                if (locked2) {
                    try {
                        // 执行业务逻辑
                        return action.execute();
                    } finally {
                        warehouse2Lock.unlock();
                    }
                } else {
                    // 获取第二个锁超时，释放第一个锁，准备重试
                    throw new LockTimeoutException("超时等待第二个锁");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("线程被中断", e);
            } catch (LockTimeoutException e) {
                if (attempt == MAX_RETRY - 1) {
                    throw new RuntimeException("重试次数用尽，操作失败", e);
                }
                // 重试前稍微等待，避免疯狂重试
                try {
                    Thread.sleep(10);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("线程被中断", ie);
                }
            } finally {
                // 确保所有锁都被释放
                if (locked2) {
                    warehouse2Lock.unlock();
                }
                if (locked1) {
                    warehouse1Lock.unlock();
                }
            }
        }
        throw new RuntimeException("不应到达这里");
    }
    
    // 函数式接口，用于执行业务逻辑
    @FunctionalInterface
    public interface RunnableWithResult<T> {
        T execute() throws Exception;
    }
    
    private static class LockTimeoutException extends RuntimeException {
        public LockTimeoutException(String message) {
            super(message);
        }
    }
}

这套机制的精妙之处

统一顺序：仍然先锁warehouse1，再锁warehouse2，从根本上避免死锁
超时保护：如果第二个锁拿不到，不会无限等待
回滚机制：超时后立即释放已持有的锁，让其他线程有机会执行
重试策略：失败后稍作等待再重试，给系统喘息的机会
最大重试次数：防止无限重试导致资源浪费

老王的完整解决方案

老王在被罚50万之后，痛定思痛，设计了一套完整的防死锁库存系统：

import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.Map;
import java.util.HashMap;
import java.util.List;
import java.util.ArrayList;
import java.util.Comparator;

/**
 * 老王版的防死锁库存系统
 * 核心思想：统一加锁顺序 + 超时重试 + 事务回滚
 */
public class WangsDeadlockFreeInventory {
    
    // 每个仓库的锁
    private final Map<Integer, ReentrantLock> lockMap = new HashMap<>();
    // 每个仓库的库存
    private final Map<Integer, Integer> inventory = new HashMap<>();
    
    private static final int MAX_RETRY = 5;
    private static final long LOCK_TIMEOUT_MS = 50;
    private static final long RETRY_DELAY_MS = 10;
    
    public WangsDeadlockFreeInventory(List<Integer> warehouseIds) {
        for (Integer id : warehouseIds) {
            lockMap.put(id, new ReentrantLock());
            inventory.put(id, 1000); // 初始库存
        }
    }
    
    /**
     * 获取锁的排序列表（按仓库ID排序，确保全局一致）
     */
    private List<ReentrantLock> getSortedLocks(Integer... warehouseIds) {
        List<ReentrantLock> locks = new ArrayList<>();
        for (Integer id : warehouseIds) {
            locks.add(lockMap.get(id));
        }
        // 按仓库ID排序，确保所有线程加锁顺序一致
        locks.sort(Comparator.comparingInt(l -> getWarehouseId(l)));
        return locks;
    }
    
    private int getWarehouseId(ReentrantLock lock) {
        for (Map.Entry<Integer, ReentrantLock> entry : lockMap.entrySet()) {
            if (entry.getValue() == lock) {
                return entry.getKey();
            }
        }
        return Integer.MAX_VALUE;
    }
    
    /**
     * 转账操作：从fromWarehouse转到toWarehouse
     */
    public boolean transfer(Integer fromWarehouse, Integer toWarehouse, int amount) {
        if (fromWarehouse.equals(toWarehouse)) {
            return false;
        }
        
        for (int attempt = 0; attempt < MAX_RETRY; attempt++) {
            List<ReentrantLock> locks = getSortedLocks(fromWarehouse, toWarehouse);
            ReentrantLock firstLock = locks.get(0);
            ReentrantLock secondLock = locks.get(1);
            
            boolean firstLocked = false;
            boolean secondLocked = false;
            
            try {
                // 按顺序加第一个锁
                firstLocked = tryLockWithInterrupt(firstLock);
                if (!firstLocked) {
                    continue;
                }
                
                // 尝试加第二个锁，带超时
                secondLocked = secondLock.tryLock(LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
                if (!secondLocked) {
                    // 超时，释放第一个锁，准备重试
                    continue;
                }
                
                try {
                    // 执行转账逻辑
                    performTransfer(fromWarehouse, toWarehouse, amount);
                    return true;
                } finally {
                    if (secondLocked) {
                        secondLock.unlock();
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            } finally {
                if (firstLocked) {
                    firstLock.unlock();
                }
            }
        }
        
        // 所有重试都失败
        return false;
    }
    
    /**
     * 执行实际的库存变更
     */
    private void performTransfer(Integer from, Integer to, int amount) {
        int fromStock = inventory.getOrDefault(from, 0);
        if (fromStock < amount) {
            throw new IllegalArgumentException("库存不足");
        }
        inventory.merge(from, -amount, Integer::sum);
        inventory.merge(to, amount, Integer::sum);
    }
    
    /**
     * 可中断的加锁
     */
    private boolean tryLockWithInterrupt(ReentrantLock lock) throws InterruptedException {
        return lock.tryLock();
    }
    
    /**
     * 批量操作：同时对多个仓库进行操作
     */
    public boolean batchUpdate(Map<Integer, Integer> updates) {
        if (updates.isEmpty()) {
            return true;
        }
        
        List<Integer> sortedIds = updates.keySet().stream()
            .sorted()
            .toList();
        
        List<ReentrantLock> locks = sortedIds.stream()
            .map(id -> lockMap.get(id))
            .toList();
        
        for (ReentrantLock lock : locks) {
            lock.lock();
        }
        
        try {
            for (Map.Entry<Integer, Integer> entry : updates.entrySet()) {
                Integer id = entry.getKey();
                Integer delta = entry.getValue();
                inventory.merge(id, delta, Integer::sum);
            }
            return true;
        } finally {
            // 逆序解锁
            for (int i = locks.size() - 1; i >= 0; i--) {
                locks.get(i).unlock();
            }
        }
    }
    
    // 测试代码
    public static void main(String[] args) throws InterruptedException {
        WangsDeadlockFreeInventory system = new WangsDeadlockFreeInventory(List.of(1, 2, 3));
        
        ExecutorService executor = Executors.newFixedThreadPool(10);
        CountDownLatch latch = new CountDownLatch(100);
        
        // 模拟100个并发转账操作
        for (int i = 0; i < 100; i++) {
            final int from = (i % 3) + 1;
            final int to = ((i + 1) % 3) + 1;
            
            executor.submit(() -> {
                try {
                    boolean success = system.transfer(from, to, 10);
                    System.out.println(Thread.currentThread().getName() + 
                        " 从仓库" + from + "转" + (success ? "成功" : "失败") + "到仓库" + to);
                } finally {
                    latch.countDown();
                }
            });
        }
        
        latch.await(10, TimeUnit.SECONDS);
        executor.shutdown();
        
        System.out.println("最终库存状态: " + system.inventory);
    }
}

为什么这套方案能彻底解决问题？

1. 打破循环等待

统一加锁顺序意味着：所有线程都按照相同的顺序获取锁。如果线程A持有锁1，等待锁2；线程B也持有锁1（在等待），它不可能持有锁2去等待锁1。循环等待被彻底打破。

2. 超时机制防止资源耗尽

即使有极小概率出现锁竞争加剧，超时机制也能确保线程不会无限等待。释放锁之后，其他线程可以继续执行，系统不会卡死。

3. 重试策略提高成功率

短暂的等待后重试，给竞争线程释放锁的机会。大多数情况下，重试1-2次就能成功。

4. 可中断的锁获取

使用tryLock而非lock，配合线程中断检查，确保线程在需要时能够及时退出等待状态。

老王的反思：从50万中学到的教训

老王后来在团队内部分享时说：

“死锁不是bug，它是设计缺陷。你永远无法通过’测试’来发现所有死锁场景，因为死锁的发生依赖于巧合的时间点。”

他总结了三个原则：

永远不要相信运气：假设死锁一定会发生，然后从设计上杜绝它
加锁顺序是底线：这是最简单、最有效的防死锁手段
超时重试是保险：即使顺序错了，也能及时发现并恢复

他还特意强调了一个细节：

“很多人以为tryLock不

正文

老王因锁顺序混乱死锁赔了50万用统一加锁顺序与超时重试彻底解决悲观锁死锁难题

死锁：程序员最讨厌的”幽灵”

老王当年的”血泪代码”

解决方案一：统一加锁顺序

为什么统一顺序能解决问题？

解决方案二：超时重试机制

这套机制的精妙之处

老王的完整解决方案

为什么这套方案能彻底解决问题？

1. 打破循环等待

2. 超时机制防止资源耗尽

3. 重试策略提高成功率

4. 可中断的锁获取

老王的反思：从50万中学到的教训

相关阅读

破解悲观锁死锁难题：掌握五大实用技巧，让数据库运行更高效

掌握悲观锁，破解死锁难题：五大实战技巧助你轻松应对数据库并发控制挑战

员工抢会议室引发死锁？用悲观锁加超时重试轻松破解并发冲突实战指南

如何巧妙使用悲观锁避免数据库死锁问题

学会悲观锁，轻松破解数据库死锁难题

从银行转账死锁到电商订单卡单悲观锁常见陷阱与防死锁实战技巧

同事加锁忘释放导致系统卡死如何用悲观锁和规范流程预防死锁详解

如何巧妙使用悲观锁避免数据库死锁问题解析

程序员锁死数据库怎么办用悲观锁防死锁实战技巧分享

如何运用悲观锁巧妙避免数据库死锁问题详解