目录

从现实世界寻找答案

没有免费的午餐

破坏占有且等待条件

破坏不可抢占条件

破坏循环等待条件

总结


在上一篇文章中,我们用 Account.class 作为互斥锁,来解决银行间的转账问题,虽然这个方案不存在并发问题,但是所有的账户的转账都是串行的,这种转账串行化性能太差。这里我们尝试把性能提升一下。

 

从现实世界寻找答案

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

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

 

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

05 - 线程死锁了,怎么办?

代码实现如下:

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;
        }
      }
    }
  } 
}

 

没有免费的午餐

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

 

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

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

05 - 线程死锁了,怎么办?

 

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;
        }
      }
    }
  } 
}

 

放到线程中,线程A和线程B会在1 和 2 出无限等待下去,因而死锁。

 

如何预防死锁

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

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

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

 

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

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

 

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

 

破坏占有且等待条件

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

05 - 线程死锁了,怎么办?

对于代码如下:

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)
    }
  } 
}

 

破坏不可抢占条件

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

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

 

破坏循环等待条件

破坏循环等待这个条件,需要对资源进行排序,然后按序申请资源。这个实现非常简单,我们假设每个账户都有不同的属性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;
        }
      }
    }
  } 
}

 

总结

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

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

相关文章: