线程中的死锁

一、死锁

死锁造成的影响很少会立即显现出来。如果一个类可能发生死锁，那么并不意味着每次都会发生死锁，而只是表示有可能，当死锁出现时，往往是在最糟糕的时候——高负载情况下。

1、锁顺序死锁

我们使用加锁来避免线程安全，但如果过度的使用加锁，则可能导致锁顺序死锁(Lock-Ordering-Deadlock)。

public class LeftRightDeadLock {
    private final Object left = new Object();
    private final Object right = new Object();
    
    public void leftRight(){
        synchronized (left) {
            synchronized (right) {
                //doSomethoing();
            }
        }
    }
    
    public void rightLeft(){
        synchronized (right) {
            synchronized (left) {
                //doSomethoing();
            }
        }
    }
}

死锁原因：

两个线程试图以不同的顺序来获得相同的锁。LeftRight线程获得left锁而尝试获得right锁，而rightLeft线程获得了right锁而尝试获得left锁，并且两个线程的操作是交错执行的，因此它们会发生死锁。

解决方法：

如果按照相同的顺序来请求锁，那么就不会发生死锁。例如，每个需要L和M的线程都一相同的顺序来获取L和M，就不会发生死锁了。

2、动态锁顺序死锁

下面的代码：将资金从一个账户转入另一个账户。在开始转账之前，首先要获得这两个Account对象的锁，以确保通过原子的方式来更新两个账户中的余额。

public void transferMoney(Account fromAccount, Account toAccount, DollarAmount amount) throws InsufficientFundsException {
    synchronized (fromAccount) {
        synchronized (toAccount) {
            if (fromAccount.getBalance().compareTo(amount) < 0) {
                fromAccount.debit(amount);
                toAccount.credit(amount);
            }
        }
    }
}

死锁原因：

所有的线程似乎都是按照相同的顺序来获得锁，但事实上锁的顺序取决于传递给transferMoney的参数顺序，而这些参数顺序又取决于外部输入。如果两个线程同时调用transferMoney，其中一个线程从X向Y转账，另一个线程从Y向X转账，那么就会发生死锁

如果执行时序不当，那么A可能获得myAccount的锁并等待yourAccount锁，而B持有yourAccount的锁并等待myAccount的锁。

解决方法：

这种死锁可以使用锁顺序死锁中的方法来检查——查看是否存在嵌套的锁获取操作。由于我们无法控制参数的顺序，因此要解决这个问题，必须定义锁的顺序，并在整个应用程序中都按照这个顺序来获取锁。①在制定锁的顺序时，可以使用System.identityHashCode()方法，该方法将返回有Object.hashCode返回的值。

private static final Object tieLock = new Object();

public void transferMoney(final Account fromAccount, final Account toAccount, final DollarAmount amount) {  
    class Helper {  
        public void transfer() {  
            if (fromAccount.getBalance().compareTo(amount) < 0) {  
                throw new RuntimeException();  
            } else {  
                fromAccount.debit(amount);  
                toAccount.credit(amount);  
            }  
        }  
    }  
   // 通过唯一hashcode来统一锁的顺序, 如果account具有唯一键, 可以采用该键来作为顺序.  
    int fromHash = System.identityHashCode(fromAccount);  
    int toHash = System.identityHashCode(toAccount);  
    if (fromHash < toHash) {  
        synchronized (fromAccount) {  
            synchronized (toAccount) {  
                new Helper().transfer();  
            }  
        }  
    } else if (fromHash > toHash) {  
        synchronized (toAccount) {  
            synchronized (fromAccount) {  
                new Helper().transfer();  
            }  
        }  
    } else {  
        synchronized (tieLock) { // 针对fromAccount和toAccount具有相同的hashcode  
            synchronized (fromAccount) {  
                synchronized (toAccount) {  
                    new Helper().transfer();  
                }  
            }  
        }  
    }  
}

②在极少数情况下，两个对象可能拥有相同的散列值，此时必须通过某种任意的方法来决定锁的顺序，而这可能又会引入死锁。为了避免这种情况，可以使用“加时赛”锁。在获得两个Account之前，首先获得这个“加时赛”锁，从而保证每次只有一个线程以未知的顺序获得这两个锁，从而消除了死锁发生的可能性（只要一致地使用这种机制）。如果经常会出现散列冲突的情况，那么这种技术可能会成为并发性的一个瓶颈（这类似于在整个程序中只有一个锁的情况），但由于System.identityHashCode中出现散列冲突的频率非常低，因此这项技术以最小的代价，换来了最大的安全性。

　　如果在Account中包含一个唯一的，不可变的并且具备可比性的键值，例如账号，那么要制定锁的顺序就更加容易了，通过键值对对象进行排序，因而不需要使用“加时赛”锁。

3、在协作对象间发生死锁

public class Taxi {
    private final Dispatcher dispatcher;
    private Point location, destination;
 
    public Taxi(Dispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }
 
    public synchronized Point getLocation() {
        return location;
    }
 
    public synchronized void setLocation(Point location){//加锁
        this.location = location;
        if(location.equals(destination)){
            dispatcher.notifyAvaliable(this);//加锁
        }
    }
}
 
public class Dispatcher {
    private final Set<Taxi> taxis;
    private final Set<Taxi> avaliableTaxis;
 
    public Dispatcher() {
        taxis = new HashSet<Taxi>();
        avaliableTaxis = new HashSet<Taxi>();
    }
 
    public synchronized void notifyAvaliable(Taxi taxi) {//加锁
        avaliableTaxis.add(taxi);//加锁
    }
 
    public synchronized Image getImage() {
        Image image = new Image();
        for (Taxi t : taxis) {
            image.drawMarker(t.getLocation());
        }
        return image;
    }
}

死锁原因：

尽管没有任何方法会显式的获取两个锁，但setLocation和getImage等方法的调用者都会获得两个锁。因为setLocation和notifyAvailable都是同步方法，因此调用setLocation的线程将首先获得Taxi的锁，然后获取Dispatcher的锁，同样调用getImage的线程将首先获取Dispatcher的锁，然后再获取每一个Taxi的锁，两个线程按照不同的顺序来获取两个锁，这时就有可能产生死锁。

解决方案：

开放调用（如果在调用某个方法时不需要持有锁，那么这种调用被称为开放调用），使同步代码块仅被用于保护那些涉及共享状态的操作

public class Taxi {
    private final Dispatcher dispatcher;
    private Point location, destination;
 
    public Taxi(Dispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }
 
    public synchronized Point getLocation() {
        return location;
    }
 
    public synchronized void setLocation(Point location) {
        boolean reachedLocation;
        synchronized (this) {
            this.location = location;
            reachedLocation = location.equals(destination);
        }
        if (reachedLocation) {
            dispatcher.notifyAvaliable(this);
        }
    }
}
 
public class Dispatcher {
    private final Set<Taxi> taxis;
    private final Set<Taxi> avaliableTaxis;
 
    public Dispatcher() {
        taxis = new HashSet<Taxi>();
        avaliableTaxis = new HashSet<Taxi>();
    }
 
    public synchronized void notifyAvaliable(Taxi taxi) {
        avaliableTaxis.add(taxi);
    }
 
    public Image getImage(){
        Set<Taxi> copy;
        synchronized (this){
            copy = new HashSet<Taxi>();
        }
        Image image = new Image();
        for(Taxi t: copy){
            image.drawMarker(t.getLocation());
        }
        return image;
    }
}

4、资源死锁

二、死锁的避免和诊断

如果必须获取多个锁，那么在设计时必须考虑锁的顺序：尽量减少潜在的加锁交互数量，将获取锁时需要遵循的协议写入正式文档并始终遵循这些协议。

在使用细粒度锁的程序中，可以通过使用一种两阶段策略(Two-Part-Strategy)来检查代码中的死锁：

首先，找出在什么地方将获取多个锁(使这个集合尽量小)，然后对所有这些实例进行全局分析，从而确保它们在整个程序中获取锁的顺序都保持一致。尽可能使用开放调用，这能极大地简化分析过程。如果所有的调用都是开放调用，那么要发现获取多个锁的实例是非常简单的，可以通过代码审查，或借助自动化的源码分析工具。

1.使用定时锁

还有一项技术可以检测死锁和从死锁中恢复过来，即显式使用Lock类中的定时tryLock功能来代替内置锁机制。

2.通过线程转储信息来分析死锁

JVM可以通过线程转储(Thread Dump)来帮助识别死锁的发生。

步骤可以参考使用jstack排查线程问题 ，只需将文中的程序替换成会导致死锁程序即可。

三、其它活跃性危险

1.饥饿

所谓“饥饿”指的是线程因无法访问所需资源而无法执行下去的情况。引发饥饿最常见的最常见资源就是CPU始终时间周期。如果线程优先级“不均”，在 CPU 繁忙的情况下，优先级低的线程得到执行的机会很小，就可能发生线程“饥饿”；持有锁的线程，如果执行的时间过长，也可能导致“饥饿”问题。
解决“饥饿”问题的方案很简单，有三种方案：

一是保证资源充足，二是公平地分配资源，三就是避免持有锁的线程长时间执行。

这三个方案中，方案一和方案三的适用场景比较有限，因为很多场景下，资源的稀缺性是没办法解决的，持有锁的线程执行的时间也很难缩短。倒是方案二的适用场景相对来说更多一些。那如何公平地分配资源呢？在并发编程里，主要是使用公平锁。也就是一种先来后到，线程的等待是有顺序的，排在等待队列前面的线程会优先获得资源。

2.活锁

活锁(Livelock) 是另一种形式的活跃性问题：它不会阻塞线程，但也不能继续执行，因为线程将不断重复执行相同的操作，而且总是失败。可以类比现实世界里的例子，路人甲从左手边出门，路人乙从右手边进门，两人为了不相撞而互相谦让，路人甲让路走右手边，路人乙也让路走左手边，结果是两人又相撞了。这种情况，基本上谦让几次就解决了，因为人会交流。可是如果这种情况发生在程序中，就有可能会一直没完没了地“谦让”下去，成为没有发生阻塞但依然执行不下去的“活锁”。解决“活锁”的方案很简单，谦让时，尝试等待一个随机的时间就可以了。例如上面的那个例子，路人甲走左手边发现前面有人，并不是立刻换到右手边，而是等待一个随机的时间后，再换到右手边；同样，路人乙也不是立刻切换路线，也是等待一个随机的时间再切换。由于路人甲和路人乙等待的时间是随机的，所以同时相撞后再次相撞的概率就很低了。“等待一个随机时间”的方案虽然很简单，却非常有效，Raft 这样知名的分布式一致性算法中也用到了它。

活锁通常发生在处理事务消息的应用程序中：如果不能成功地处理某个消息，那么消息处理机制将会回滚整个事务，并将它重新放到队列的开头。如果消息处理器在处理某种特定类型的消息时存在错误并导致它失败，那么每当这个消息从队列中取出并传递到存在错误的处理器时，都会发生事务回滚。由于这条消息被放到队列开头，因此消息处理器将被反复调用，并返回相同的结果(有时也被称为毒药消息，Poison Message)。虽然处理消息的线程没阻塞，但也无法继续执行。这种形式的活锁通常是由过度的错误恢复代码造成的，因为它错误地将不可修复的错误作为可修复的错误。

3.糟糕的响应性(性能问题)

例如GUI应用程序中，如果使用了后台线程，而后台线程执行CPU密集型的任务，则可能导致用户界面失去响应。

锁的过度使用可能导致串行化的范围过大，这样就不能够发挥多线程的优势了，而我们之所以使用多线程搞并发程序，为的就是提升性能。

怎么才能避免锁带来的性能问题呢？这个问题很复杂，从方案层面来讲，可以这样解决：

第一，既然使用锁会带来性能问题，那最好的方案自然就是使用无锁的算法和数据结构了。

在这方面有很多相关的技术，例如线程本地存储 、写入时复制 (Copy-on-write)、乐观锁等；Java 并发包里面的原子类也是一种无锁的数据结构；Disruptor 则是一个无锁的内存队列，性能都非常好。
第二，减少锁持有的时间。互斥锁本质上是将并行的程序串行化，所以要增加并行度，一定要减少持有锁的时间。

这个方案具体的实现技术也有很多，例如使用细粒度的锁，一个典型的例子就是 Java 并发包里的 ConcurrentHashMap，它使用了所谓分段锁的技术（这个技术后面我们会详细介绍）；还可以使用读写锁，也就是读是无锁的，只有写的时候才会互斥。

总结：

1.死锁的几种形式，以及各自的解决方法？

①.锁顺序死锁

　　按照相同的顺序依次获取锁。

②.动态的锁顺序死锁

　　按照相同的顺序依次获取锁。

③.在协作对象间发生的死锁

　　开放调用

④.资源死锁：线程相互等待对方的锁、线程饥饿死锁

2.如何预防死锁？

①.按照相同的加锁顺序进行获取锁。

②.使用两阶段策略来检查代码中的死锁。(尽可能使用开放调用，使用代码审查或借助源码分析工具来分析)

③.使用定时锁

3.发生死锁后如何诊断死锁？

使用JVM转储分析死锁

4.除了死锁，活跃性问题还有哪些？如何解决？

①死锁 

②活锁

③饥饿

④糟糕的响应性

参考资料：

《Java并发编程实战》第10章