一个简单的转账场景示例带你了解并发安全？

本文转帐场景主要参考来自于极客时间 王老师的 《Java 并发编程实战》

一个简单的转账场景示例带你了解并发安全？

例如如银行业务里面的转账操作，账户 A 减少 100 元，账户 B 增加 100 元。
我们声明了个账户类：Account，该类有一个成员变量余额：balance，还有一个用于转账的方法：transfer()，然后怎么保证转账操作 transfer() 没有并发问题呢？

示例代码如下：

classAccount{privateint balance;// 转账voidtransfer(Account target,int amt){if(this.balance > amt){this.balance -= amt;
      target.balance += amt;}}}

首先直觉告诉我们，有线程安全问题那就用 synchronized 关键字修饰一下 transfer() 方法不就可以了，如下所示。

classAccount{privateint balance;// 转账synchronizedvoidtransfer(Account target,int amt){if(this.balance > amt){this.balance -= amt;
      target.balance += amt;}}}

在这段代码中，问题出在哪里呢？问题就出在 this 这把锁上，this 这把锁可以保护自己的余额 this.balance，却保护不了别人的余额 target.balance。
具体可以分析一下，假设有 A、B、C 三个账户，余额都是 200 元，我们用两个线程分别执行两个转账操作：账户 A 转给账户 B 100 元，账户 B 转给账户 C 100 元，最后我们期望的结果应该是账户 A 的余额是 100 元，账户 B 的余额是 200 元， 账户 C 的余额是 300 元。

实际多线程执行结果可能为最终账户 B 的余额可能是 300，可能是 100，自行分析或验证。

使用锁的正确姿势

this 是对象级别的锁，所以 A 对象和 B 对象都有自己的锁，我们只要让 A 对象和 B 对象共享一把锁，那就能解决并发安全问题。
我们于是可以用 Account.class 作为共享的锁。这个对象是 Java 虚拟机在加载 Account 类的时候创建的，所以我们不用担心它的唯一性。使用 Account.class 作为共享的锁，代码修正示例如下：

classAccount{privateint balance;// 转账voidtransfer(Account target,int amt){synchronized(Account.class){if(this.balance > amt){this.balance -= amt;
        target.balance += amt;}}}}

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

向现实世界要答案

现实世界里，账户转账操作是支持并发的，而且绝对是真正的并行，我们试想在古代，没有信息化，账户的存在形式真的就是一个账本，而且每个账户都有一个账本，这些账本都统一存放在文件架上。银行柜员在给我们做转账时，要去文件架上把转出账本和转入账本都拿到手，然后做转账。

上面这个过程在编程的世界里怎么实现呢？其实用两把锁就实现了，转出账本一把，转入账本另一把。在 transfer() 方法内部，我们首先尝试锁定转出账户 this（先把转出账本拿到手），然后尝试锁定转入账户 target（再把转入账本拿到手），只有当两者都成功时，才执行转账操作。

classAccount{privateint balance;// 转账voidtransfer(Account target,int amt){// 锁定转出账户synchronized(this){// 锁定转入账户synchronized(target){if(this.balance > amt){this.balance -= amt;
          target.balance += amt;}}}}}

我们知道，使用细粒度锁可以提高并行度，但是也可能会导致死锁。

如何预防死锁？

并发程序一旦死锁，一般没有特别好的方法，很多时候我们只能重启应用。因此，解决死锁问题最好的办法还是规避死锁。
要避免死锁就需要分析死锁发生的条件，有个叫 Coffman 的牛人早就总结过了，只有以下这四个条件都发生时才会出现死锁：

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

所以，我们只要破坏其中一个条件，就能成功避免死锁的发生。
其中，互斥这个条件我们没有办法破坏，因为我们用锁为的就是互斥。不过其他三个条件都是有办法破坏掉的，到底如何做呢？

破坏不可抢占条件

java.util.concurrent 这个包下面提供的 Lock 是可以轻松解决这个问题的。

破坏循环等待条件

破坏这个条件，需要对资源进行排序，然后按序申请资源。这个实现非常简单，我们假设每个账户都有不同的属性 id，这个 id 可以作为排序字段，申请的时候，我们可以按照从小到大的顺序来申请。示例代码如下：

classAccount{privateint id;privateint balance;// 转账voidtransfer(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;}}}}}

破坏占用且等待条件

从理论上讲，要破坏这个条件，可以一次性申请所有资源。示例代码如下：

classAllocator{privateList<Object> als =newArrayList<>();// 一次性申请所有资源synchronizedbooleanapply(Object from,Objectto){if(als.contains(from)||
         als.contains(to)){returnfalse;}else{
      als.add(from);
      als.add(to);}returntrue;}// 归还资源synchronizedvoidfree(Object from,Objectto){
    als.remove(from);
    als.remove(to);}}classAccount{// actr应该为单例privateAllocator actr;privateint balance;// 转账voidtransfer(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)}}}

上面用死循环的方式实现等待有什么问题呢？

如果 apply() 操作耗时非常短，而且并发冲突量也不大时，这个方案还挺不错的，因为这种场景下，循环上几次或者几十次就能一次性获取转出账户和转入账户了。但是如果 apply() 操作耗时长，或者并发冲突量大的时候，循环等待这种方案就不适用了，因为在这种场景下，可能要循环上万次才能获取到锁，太消耗 CPU 了。

其实在这种场景下，最好的方案应该是：如果线程要求的条件不满足，则线程阻塞自己，进入等待状态；当线程要求的条件满足后，通知等待的线程重新执行。其中，使用线程阻塞的方式就能避免循环等待消耗 CPU 的问题。所以我们可以用等待通知机制来优化此流程，示例代码如下：

classAllocator{privateList<Object> als;// 一次性申请所有资源synchronizedvoidapply(Object from,Objectto){// 经典写法while(als.contains(from)||
         als.contains(to)){try{wait();}catch(Exception e){}} 
    als.add(from);
    als.add(to);}// 归还资源synchronizedvoidfree(Object from,Objectto){
    als.remove(from);
    als.remove(to);notifyAll();}}

• notify() 是会随机地通知等待队列中的一个线程；
• notifyAll() 会通知等待队列中的所有线程，推荐尽量使用 notifyAll()

并发编程全景图

个人总结及归纳的思维导图，供大家参考。