如何处理线程死锁

在上一篇文章中，我们用 Account.class 作为互斥锁，来解决银行间的转账问题，虽然这个方案不存在并发问题，但是所有的账户的转账都是串行的，例如账户 A 转账户 B、账户 C 转账户 D 这两个转账操作现实世界里是可以并行的，但是在这个方案里却被串行化了，这样的话，性能太差。

那下面我们就尝试着把性能提升一下。

1. 从现实世界寻找答案

现实世界里，账户转账操作是支持并发的，而且绝对是真正的并发。我们先试想一下古代，账户的存在形式就是一个账本，而且每个账户都有一个账本，统一放在文件架上，在做转账时，去文件架上把转出和转入账本都拿到手，然后转账，那么，这个柜员在拿账本的时候可能遇到以下三种情况：

1. 文件架上恰好有转出和转入账本，同时拿走
2. 文件架上只有转出或转入账本，那柜员就先拿走有的账本，同时等着其他柜员把另一个账本送回
3. 转出和转入账本都没有，柜员等着两个账本都被送回来。

上面的过程在编程的世界里怎么实现呢？其实用两把锁就实现了，转出账本一把，转入账本一把。在 transfer() 方法内部，我们首先尝试锁定转出账户 this，然后在尝试锁定转入账户 target，只有当两者都成功了，才执行转账操作，过程如下图所示：

代码实现如下：

class Account {
    private int balance;
 
    // 转账
    void transfer(Account target, int amt) {
        // 锁定转出账户
        synchronized (this) {
            // 锁定转入账户
            synchronized (target) {
                if (this.balance > amt) {
                    this.balance -= amt;
                    target.balance += amt;
                }
            }
        }
    }
}

2. 没有免费的午餐

上面的实现看上去很完美，相对于使用 Account.class 作为互斥锁，锁定的范围太大，而我们锁定两个账户范围小多了，这样的锁叫细粒度锁。使用细粒度锁可以提高并行度，是性能优化的一个重要手段。但是，使用细粒度锁是有代价的，这个代价就是可能会导致死锁。

那么什么是死锁呢？死锁的定义是：一组相互竞争资源的线程因相互等待，导致永久阻塞的现象。

上面的代码我们可以举个现实中的例子。如果有客户找柜员张三做个转账：账户 A 转到 账户 B 100元，此时另一个客户找柜员李四也做个转账业务：账户B 转账户 A 100元，于是张三和李四同时都去文件架上拿账本，此时凑齐张三拿到了账户A，李四拿到了账户B ，张三拿着账户A 后就等着账户B ，而李四拿到账户B后就等着账户A，它们会永远的等下去，因为张三不会把账户A送回去，李四也不会把账户B 送回去，因此就死等下去。

class Account {
    private int balance;
 
    // 转账
    void transfer(Account target, int amt) {
        // 锁定转出账户
        synchronized (this) {     ①
            // 锁定转入账户
            synchronized (target) { ②
                if (this.balance > amt) {
                    this.balance -= amt;
                    target.balance += amt;
                }
            }
        }
    }
}

上面转账的代码是怎么发生死锁的呢？我们假设线程 T1 执行账户 A 转账户 B 的操作，账户 A.transfer(账户 B)；同时线程 T2 执行账户 B 转账户 A 的操作，账户 B.transfer(账户 A)。当 T1 和 T2 同时执行完①处的代码时，T1 获得了账户 A 的锁（对于 T1，this 是账户 A），而 T2 获得了账户 B 的锁（对于 T2，this 是账户 B）。之后 T1 和 T2 在执行②处的代码时，T1 试图获取账户 B 的锁时，发现账户 B 已经被锁定（被 T2 锁定），所以 T1 开始等待；T2 则试图获取账户 A 的锁时，发现账户 A 已经被锁定（被 T1 锁定），所以 T2 也开始等待。于是 T1 和 T2 会无期限地等待下去，也就是我们所说的死锁了。

 

3. 如何预防死锁

并发程序一旦死锁，一般没有好的方法，很多时候我们只能重启应用。因此，解决死锁问题做好的办法就是规避死锁。

要避免死锁就要分析出死锁发生的条件，前人总结出，发生以下四个条件是才会出现死锁：

1. 互斥，共享资源 X 和 Y 只能被一个线程占用；
2. 占有且等待，线程 T1 已经取得共享资源 X，在等待共享资源 Y 的时候，不释放共享资源 X；
3. 不可抢占，其它线程不能强行抢占线程 T1 占有的资源；
4. 循环等待，线程 T1 等待线程 T2 占有的资源，线程 T2 等待线程 T1 占有的资源。

反过来分析，也就是说我们只要破坏其中一个，就可以成功避免死锁的发生。

1. 破坏互斥：互斥这个条件我们没有办法破坏，因为我们用锁就是为了互斥；
2. 破坏占有且等待：我们可以一次性申请所有的资源，这样就不存在等待了；
3. 破坏不可抢占：占有部分这样的线程进一步申请其他资源时，如果申请不到，可以主动释放占有的资源；
4. 破坏循环等待：可以靠按序申请资源来预防。所谓按序申请，是指资源是有线性顺序的，申请的时候可以先申请资源序号小的，在申请资源序号大的，资源线性化后自然就不存在循环了。

我们已经从理论上解决了如何预防死锁，那具体如何体现在代码上，下面我们就来尝试用代码实践以下这些理论。

3.1 破坏占有且等待条件

从理论上讲，要破坏这个条件，可以一次性申请所有资源。在现实世界里，对于转账操作来说，它需要的资源有两个，一个是转出账户，一个是转入账户，怎样同时申请这两个账户呢？方法是，可以增加一个账本管理员，然后只允许账本管理员从文件架上拿账本，例如张三同时申请账本A 和 B，只有当账本 A 和 B 都在的时候才会给张三，只有就保证了一次性申请所有只有。

class Allocator {
    private List<Object> als = new ArrayList<>();
    // 一次性申请所有资源
    synchronized boolean apply(
            Object from, Object to) {
        if (als.contains(from) ||
                als.contains(to)) {
            return false;
        } else {
            als.add(from);
            als.add(to);
        }
        return true;
    }
    // 归还资源
    synchronized void free(
            Object from, Object to) {
        als.remove(from);
        als.remove(to);
    }
}
 
class Account {
    // actr 应该为单例
    private Allocator actr;
    private int balance;
    // 转账
    void transfer(Account target, int amt) {
        // 一次性申请转出账户和转入账户，直到成功
        while (!actr.apply(this, target));
        try {
            // 锁定转出账户
            synchronized (this) {
                // 锁定转入账户
                synchronized (target) {
                    if (this.balance > amt) {
                        this.balance -= amt;
                        target.balance += amt;
                    }
                }
            }
        } finally {
            actr.free(this, target);
        }
    }
}

3.2 破坏不可抢占条件

破坏不可抢占资源条件看上去很简单，核心是要能够主动释放它占有的资源，这一点 synchronized 做不到，原因是 synchronized 申请资源时，如果申请不到，线程直接进入阻塞状态，而线程进入阻塞状态了，啥也干不了，也释放不了线程占有的资源。

不过 synchronized 解决不了，在Java SDK 中 java.util.concurrent 这个包下提供的 Lock 是可以轻松解决这个问题的。

3.3 破坏循环等待条件

破坏循环等待这个条件，需要对资源进行排序，然后按序申请资源。这个实现非常简单，我们假设每个账户都有不同的属性ID，按个ID 可以作为排序字断，申请的时候，我们可以按照从小到大的顺序来申请。

class Account {
    private int id;
    private int balance;
 
    // 转账
    void transfer(Account target, int amt) {
        Account left = this        ①
        Account right = target;    ②
        if (this.id > target.id) { ③
            left = target;           ④
            right = this;            ⑤
        }                          ⑥
        // 锁定序号小的账户
        synchronized (left) {
            // 锁定序号大的账户
            synchronized (right) {
                if (this.balance > amt) {
                    this.balance -= amt;
                    target.balance += amt;
                }
            }
        }
    }
}

4. 总结

这一篇文章主要讲了用细粒度锁来锁定多个资源时，要注意死锁的问题。这个就需要你能把它强化为一个思维定势，遇到这种场景，马上想到可能存在死锁问题。当你知道风险之后，才有机会谈如何预防和避免，因此，识别出风险很重要。

用细粒度锁来锁定多个资源时，可以提高并行度，提升系统性能，但要注意死锁问题。预防死锁主要是破坏三个条件中的一个，有了这个思路，现实就简单了。但仍需注意的是，有时候预防死锁成本也是很高的。例如上面的例子，我们破坏占有且等待条件的成本就比破坏循环等待条件的成本高，破坏占有且等待体检，我们也是锁了所有的用户，而且还是用了死循环 while(!actr.apply(this, target)); 方法，不过好在apply() 方法基本不耗时。在转账这个例子中，破坏循环等待条件就是成本最低的一个方案。

所以我们在选择具体方案时，还需要评估以下操作成本，从中选择一个成本最低的方案。

 

5. 思考

我们上面提到：破坏占用且等待条件，我们也是锁了所有的账户，而且还是用了死循环 while(!actr.apply(this, target));这个方法，那它比 synchronized(Account.class) 有没有性能优势呢？

其实，synchronized(Account.class) 锁了Account类相关的所有操作。相当于文中说的包场了，只要与Account有关联，通通需要等待当前线程操作完成。while死循环的方式只锁定了当前操作的两个相关的对象。两种影响到的范围不同。