并发编程学习(5) —— 如何解决死锁

亦凉 2022-03-01 13:36 236阅读 0赞

## 前沿 ##

在[并发编程学习(4) —— 互斥锁如何保护多个资源][4_ _]中讲述了入好保护关联资源以及不关联资源，但是里面的方法并不是最好，要想性能提高就要优化，但同样也会带来优化后的问题，接下来慢慢讨论。

## 性能差的原因 ##

为了方便阅读，我这里贴上上一篇文章的保护关联资源的代码：

public class Account { 
        private int balance; // 余额
        // 转账
        void transfer(Account account, int money){ 
            synchronized(Account.class){ 
                if(this.balance > money){ 
                    account.balance += money;
                    this.balance -= money;
                }
            }
        }
    }

可以看到，余额资源被Account.class的锁保护，但是这里有个很大的问题，用Account.class作为锁的话，就会使整个转账操作串行话，就是说如果出现线程1执行A转给B，线程2执行C转给D的时候，线程2必须要等到线程线程1结束才能执行。这意味着什么？如果一天有500万人（这还是最少的人数）进行转账，假设1秒转账完成一次，要等多久才能把这些500万人的钱全部转账完成？更何况还会有新增的，这样的性能很明显是不能接受的，我们需要做的是把串行改为并行。

## 如何优化 ##

在网络还不发达的时候，人们转账的话需要到钱庄进行操作，柜员会把转出人A的账本和转入人B的账本从账本柜中取出，然后进行转账。柜员拿账本可能会出现一下三种情况：

> 1.  两本账本的其中一本账本正在被使用，那么只能等待账本使用完成并归还后才能进行转账。
> 2.  两本账本都在被使用，需要等到两边账本归还后才能转账。
> 3.  两边账本都没有被使用，那么直接取出并进行转账。

那么转换成代码就是这样：

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

现在来解释下以上的代码，首先先尝试锁定转出账户（先拿到转出账本），然后再尝试拿到转入账户(再把转入账本拿到手)，当两者成功时才能进行转账操作。这样就能缩小作用范围，这样的锁，我之前介绍过，叫**细粒度锁**。

## 使用细粒度锁可能带来的死锁问题 ##

细粒度锁是一个重要的优化手段，但同样也会带来死锁问题，那么什么是死锁？我们用以上的例子来说明，有一个客户A找柜员1进行转账业务：客户A转100元到账户B。但同时，也有另一个客户B找柜员2进行转账业务：客户B转100元到客户A。这时候柜员1和柜员2分别拿到转出账本A和转出账本B，柜员1等待账本B的归还，而柜员2则等待账本A的归还。在账本B没有归还前，柜员1不会把账本A送回去，同理柜员2，他们只会永远地等待下去，这种情况在编程领域就称为**死锁**。

**一组互相竞争资源的线程因互相等待，导致永久阻塞的现象称为死锁**，如下图所示：  
![资源发生死锁的资源分配图][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODE2ODk0Nw_size_16_color_FFFFFF_t_70]

## 如何预防死锁 ##

死锁出现一般都没什么好办法，只能重启应用，那么比起解决死锁，更好的方法是避免死锁，那么什么情况下会产生死锁？只有下列四个条件都发生时才会出现死锁：

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

那么换句话说，避免死锁，只需要破坏其中一个条件就可以了。互斥我们是无法解决的，因为我们用锁就是为了能够实现互斥。对于下列三个条件：

#### 1、占用且等待条件 ####

针对这个条件，我们可以一次性取得所需的所有资源，这样就不用等待了。还是用转账的例子来说明，我们可以在柜员和客户之间增加一个账本管理员，负责管理所有账本，柜员要取得账本必须通过账本管理员，账本管理员只会在账本A和B都在的时候才会给柜员。代码实现如下：

import java.util.ArrayList;
    import java.util.List;
    
    
    class Allocator{ 
        private List<Object> als = new ArrayList<Object>();
        
        // 一次性申请所有资源
        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);
        }
    }
    
    public class Account { 
        private int balance; // 余额
        private Allocator allocator;
    
        // 转账
        void transfer(Account account, int money){ 
            // 一次性申请资源，直到申请成功
            while (!allocator.apply(this, account));
            try{ 
                // 锁定转出账户
                synchronized (this){ 
                    // 锁定转入账户
                    synchronized(account){ 
                        if(this.balance > money){ 
                            account.balance += money;
                            this.balance -= money;
                        }
                    }
                }
            }finally { 
                allocator.free(this, account);
            }
           
        }
    }

#### 2、破坏不可抢占式条件 ####

这块内容还不太会，后续在更新。

#### 3、破坏循环等待 ####

破坏这个条件，需要对资源进行排序，按照id从小到大进行申请，从小到大锁定账户，这样就解决循环等待的问题，代码如下：

public class Account { 
        private int id;
        private int balance; // 余额
        private Allocator allocator;
    
        // 转账
        void transfer(Account account, int money){ 
            Account left = this;
            Account right = account;
            if(left.id > right.id){ 
                left = account;
                right = this;
            }
            // 锁定转出账户
            synchronized (left){ 
                // 锁定转入账户
                synchronized(right){ 
                    if(this.balance > money){ 
                        account.balance += money;
                        this.balance -= money;
                    }
                }
            }
        }
    }

这种方法貌似很完美，但是有个缺点，在并发量冲突大的情况下，可能要循环上万次的才能获取到锁，这样太消耗CPU了。如果线程能够在条件不满足的情况下进入等待状态，而条件满足时通知在等待的线程重新执行。像这种**等待-通知机制**在JAVA中是能够实现的，就是这三个方法**notifyAll()、wait()、notify()**。

## 等待-通知机制 ##

要描述等待-通知机制，我们可以通过现实的例子来进行解读：

1.患者到医院诊治（线程去获取互斥锁），当患者被叫到时，患者就能被医生诊断病因（线程已获取到互斥锁）。
    2.患者因为缺少体检报告（条件不满足），医生让患者去做体检，然后当这名患者去做体检的时候，医生叫下一个患者（线程释放互斥锁，下一个线程获取锁）。
    3.体检患者做完体检到医生处等待就诊（线程重新获取互斥锁）。

可以通过下图来增强理解：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODE2ODk0Nw_size_16_color_FFFFFF_t_70 1]

当线程需要调用资源的时候，会进入左边的等待队列来获取互斥锁，当获取锁成功后如果发现条件不满足，就会进入右边的等待队列同时释放锁，当条件满足的时候就会调用notifyAll()来通知右边的等待队列，右边等待队列的线程需要重新获取锁，然后在执行其它操作。值得一提的是，notifyAll()只能保证某一时刻的条件满足，若过了这个时间条件不满足，则线程只能继续等待。**(尽量使用notifyAll()，因为notify()只是通知等待队列中随机的一个线程，有可能导致等待队列中的线程永远不会被通知到。)**

下面是实现代码:

import java.util.ArrayList;
    import java.util.List;
    
    
    class Allocator{ 
        private List<Object> als = new ArrayList<Object>();
        
        // 一次性申请所有资源
        synchronized void apply(Object from, Object to){ 
            if(als.contains(from) || als.contains(to)){ 
    		        try{ 
    		            wait();
    		        }catch(Exception e){ }
            }else{ 
                als.add(from);
                als.add(to);
            }
        }
        
        // 归还资源
        synchronized void free(Object from, Object to){ 
            als.remove(from);
            als.remove(to);
            notifyAll();
        }
    }
    
    public class Account { 
        private int balance; // 余额
        private Allocator allocator;
    
        // 转账
        void transfer(Account account, int money){ 
            // 一次性申请资源，直到申请成功
            allocator.apply(this, account);
            try{ 
                // 锁定转出账户
                synchronized (this){ 
                    // 锁定转入账户
                    synchronized(account){ 
                        if(this.balance > money){ 
                            account.balance += money;
                            this.balance -= money;
                        }
                    }
                }
            }finally { 
                allocator.free(this, account);
            }
           
        }
    }

## 结尾 ##

其实，很多情况下都可以用现实模型来解决，但是同样也带来很多细节问题，因为人永远比机器智能，将上面的转账例子，在现实生活中柜员之间能够相互沟通来解决互相等待的过程，但在编程的世界中是缺乏这种沟通的。

[4_ _]: https://blog.csdn.net/weixin_38168947/article/details/88650820
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODE2ODk0Nw_size_16_color_FFFFFF_t_70]: /images/20220301/f697a14258334ba1a04939c519052107.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODE2ODk0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220301/0ba701b0f23a4520b27ee40fe89913c6.png