《Java并发编程实践》笔记5——线程活跃性问题及其解决方法

1.死锁的定义及发生条件：

死锁就像是两个人过独木桥，在桥中间碰到了，谁也不想让步，结果谁也无法通过。

线程A占有锁L时想要获得锁M，而线程B占有锁M并尝试获得锁L，两个线程将永远等待下去，这种情况称为死锁（deadlock），或致命拥抱（the deadly embrace）。

在并发程序设计中，死锁 (deadlock) 是一种十分常见的逻辑错误。通过采用正确的编程方式，死锁的发生不难避免。

死锁发生的四个必要条件：

(1).互斥：存在这样一种资源，它在某个时刻只能被分配给一个执行绪（也称为线程）使用；

(2).持有：当请求的资源已被占用从而导致执行绪阻塞时，资源占用者不但无需释放该资源，而且还可以继续请求更多资源；

(3).不可剥夺：执行绪获得到的互斥资源不可被强行剥夺，换句话说，只有资源占用者自己才能释放资源；

(4).环形等待：若干执行绪以不同的次序获取互斥资源，从而形成环形等待的局面，即锁顺序死锁。

上面四个条件只要打破任何一个，死锁问题就可以解决。

数据库系统通过在表示正在等待关系的有向图上搜索循环来检测死锁，一旦发现死锁会牺牲一个调用者，使它退出事务释放资源，从而使其他事务能够继续进行。

JVM中线程发送死锁时，应用系统可能完全停止或者某个特定子系统停止，也可能是性能受到影响，死锁不但经常发生在高负载情况下，而且难以测试和重现，恢复应用程序健康状态的唯一方式是重启应用。

2.动态锁顺序死锁：

线程发送死锁必要条件中的环形等待容易产生锁顺序死锁，下面例子代码演示一个简单的锁顺序死锁：

public class LeftRightDeadlock {  
    private final Object left = new Object();  
    private final Object right = new Object();  
  
    public void leftRight() {  
        synchronized (left) {  
            synchronized (right) {  
                doSomething();  
            }  
        }  
    }  
  
    public void rightLeft() {  
        synchronized (right) {  
            synchronized (left) {  
                doSomethingElse();  
            }  
        }  
    }  
}  

若一个线程调用leftRight方法，而另一个线程调用rightLeft方法，则很有可能发生死锁。

验证锁顺序一致性需要对程序中锁的行为进行整体分析，单独观察每一个锁的代码路径是不充分的，因为leftRight方法和rightLeft方法获得锁的方式都是合法的，只是它们彼此不能相互协调而已。

在实际的生产代码中，很少会出现像上面例子代码中这么明显的锁顺序死锁代码，通常会发生如下的隐式锁顺序死锁问题：

下面的转账例子演示一个动态的锁顺序死锁，代码如下：

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

若另个线程同时调用transferMoney方法，一个从X向Y转账，另一个从Y向X转账，如：

A：transferMoney(xAccount, yAccount, 10);

B：transferMoney(yAccount, xAccount, 20);

在偶然情况下很容易产生线程A获得xAccount的锁，等待yAccount的锁；线程B获得yAccount的锁，等待获取xAccount的锁，即产生锁顺序死锁。

上述隐式锁顺序死锁的例子中锁的顺序是由外部输入参数顺序决定的，解决此类锁顺序死锁的方法是确保线程以通用的固定顺序获得锁，制定锁顺序例子代码如下：

private static final Object tieLock = new Object();  
  
public void transferMoney(Account fromAccount, Account toAccount, DollarAmount amount)  
throws InsufficientFundsException{  
    class Helper{  
        public void transferMoney()throws InsufficientFundsException{  
            if(fromAccount.getBalance().compareTo(amount) < 0){  
                throw new InsufficientFundsException();  
            }else{  
                fromAccount.debit(amount);  
                toAccount.credit(amount);  
            }  
        }  
    }  
      
    int fromHash = System.identityHashCode(fromAccount);  
    int toHash = System.identityHashCode(toAccount);  
    if(fromHash < toHash){  
        synchronized(fromAccount){  
            synchronized(toAccount){  
                new Helper().transferMoney();  
            }  
        }  
    }else if(fromHash > toHash){  
        synchronized(toAccount){  
            synchronized(fromAccount){  
                new Helper().transferMoney();  
            }  
        }  
    }else{  
        synchronized(tieLock){  
            synchronized(fromAccount){  
                synchronized(toAccount){  
                    new Helper().transferMoney();  
                }  
            }  
        }  
    }  
}  

确保所有的线程以通用固定的顺序获得锁可以有效解决锁顺序死锁。

3.协作对象的锁顺序死锁：

很多情况下，锁顺序死锁发生总不像动态锁顺序死锁例子的那么明显，即很少在一个方法里面显式调用两个锁，但是同步方法的调用外部的同步方法时，同步方法上的隐式锁同样会发生锁顺序死锁问题。

以一个出租车GPS调度系统为例来演示非开放调用的锁顺序死锁问题：

public class Taxi {  
    private Point location, destination;  
    private final Dispatcher dispatcher;  
  
    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.notifyAvailable(this);  
        }  
    }  
}  
  
public class Dispatcher {  
    private final Set<Taxi> taxis;  
    private final Set<Taxi> availableTaxis;  
  
    public Dispatcher() {  
        taxis = new HashSet<Taxi>();  
        availableTaxis = new HashSet<Taxi>();  
    }  
  
    public synchronized void notifyAvailable(Taxi taxi) {  
        availableTaxis.add(taxi);  
    }  
  
    public synchronized Image getImage() {  
        Image image = new Image();  
        for (Taxi t : taxis) {  
            image.drawMarker(t.getLocation());  
        }  
        return image;  
    }  
}  

线程A：调用setLocation方法作为对GPS接收器更新的响应，则首先获得Taxi对象的锁更新位置，若已经到达目的地，则调用Dispatcher的notifyAvailable方法获取Dispatcher对象的锁来通知有出租车可用。

线程B：调用getImage方法来获取出租车的位置，则首先获取Dispatcher对象的锁，然后调用Taxi的getLocation方法获取Taxi对象的锁来获取出租车的位置。

在偶发情况线程A获取了Taxi对象的锁，等待Dispatcher对象的锁；线程B获取了Dispatcher对象的锁，等待Taxi对象的锁，产生锁顺序死锁。

上述协作对象锁顺序死锁是由非开放调用引起的，非开放调用是指在持有锁时调用外部方法，外部方法可能会获得其他锁(同步方法，产生死锁的风险)，或者遭遇严重超时的阻塞，当持有锁时会延迟其他试图获得该锁的线程。

解决协作对象的锁顺序死锁的方法是使用开放调用，开放调用是指调用方法时不需要持有锁，使用开放调用解决死锁问题的例子代码如下：

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

在程序中尽量使用开放调用，依赖于开放调用的程序，相比于那些在持有锁的时候还调用外部方法的程序，更容易进行死锁*度的分析。

4.饥饿死锁：

饥饿就像3个人打篮球，其中两个人牢牢把控着球，第三个人死活也得不到球，变成了现场观众。

饥饿死锁是指线程无法得到资源(cpu或io资源或数据库连接池资源等)，所以无法执行下去，称为饿死，比较常见的就是在优先级调度中，不停的有高优先级的线程创建，导致低优先级的线程的无法分配到cpu，从而饥饿。

饥饿死锁的例子如下：

public class ThreadDeadLock {  
    ExecutorService service = Executors.newSingleThreadExecutor();  
  
    class RenderPageTask implements Callable<String> {  
        public String call()throws Exception{  
            Future<String> header, footer;  
            header = service.submit(new LoadFileTask("header.html"));  
            footer = service.submit(new LoadFileTask("footer.html"));  
            String page = renderBody();  
            return header.get() + page + footer.get();//出现饥饿死锁，任务等待子任务结果  
        }  
    }  
}  

上述饥饿死锁的例子中，一个任务将获取页眉和页脚工作提交到单线程化的线程池中，同时等待该单线程化线程池中获取页眉和页脚任务的执行结果，最后将页眉、页脚和页面主体组合渲染页面，在该情况下，获取页眉任务等待获取页脚任务的完成，而获取页脚的任务等待获取页眉任务的完成，结果在该单线程化线程池中执行的两个任务永久停滞下来。

通常需要等待其他任务的结果(任务有依赖关系)的任务是产生线程饥饿死锁的来源，有界池和相互依赖的任务不能放在一起使用。

说起饥饿想起了曾经经历过的一个笑话，上大学的时候，有一次宿舍里面的电脑集体中毒了，尝试了很多主流的杀毒软件也没有解决，最后大家一致决定格式化硬盘重新安装操作系统，这是有个室友很淡定地关机睡觉，我们问他什么时候重装系统，他说他打算两个星期不开机饿死病毒。虽然这个笑话很冷，但也阐述了饥饿的基本原理，只不过在让病毒饥饿的同时他自己的工作也被饥饿了。

5.活锁：

活锁好比两个过于礼貌的人在半路相遇，他们都给对方让路，于是又在另一条路上相遇了，然后就这样不停的一直让路下去，导致没有人能够通过，活锁是一种特殊形式的饥饿。

活锁是尽管线程没有被阻塞，线程却仍然不能继续，因为它不断重试相同的操作，却总失败。

活锁通常发生在如下两种情况：

(1).过渡的错误恢复代码：

以消息处理为例，若消息处理失败，其中传递消息的底层架构会退回整个事务，并把它置回队首，若消息处理程序对某种特定类型消息处理存在bug，每次都会处理失败，那么每次这个消息都会被从队列中取出，传递到存在问题的消息处理器，然后发生事务回退，然后消息又被置回队首，如此反复，虽然消息处理器并没有阻塞，但是线程却永远无法继续执行下去了。

这种形式的活锁是误将不可修复的错误当做可以修复的错误来处理而导致的。

(2).多个相互协作的线程修改状态：

多个相互协作的线程为了彼此间响应而不停修改状态，使得没有一个线程能够继续(类似于让路的例子)，那么就发生了活锁。

解决活锁的一种方案是对重试机制引入随机性，例如在以太网上发送数据包时，如果发现了数据包冲突，发送的数据包的各方会随机等待一个时间继续重试，在并发程序中也类似，通过随机等待重试可以有效避免活锁。

6.死锁的避免与诊断：

前面已经介绍了死锁常见的解决方法，总结如下：

(1).定制锁顺序，使用一致的锁顺序避免锁顺序死锁。

(2).使用开放调用避免隐式锁顺序死锁。

(3).在使用显式Lock时，使用定时的tryLock特性，使得超时之后自动释放锁。

诊断死锁的方法：

当怀疑发生死锁时，强制JVM产生线程转储，通过分析线程转储分析死锁信息。