03.活跃性问题

[TOC]

概念

活跃性是指某件正确的事情最终会发生，当某个操作无法继续下去的时候，就会发生活跃性问题。在多线程中一般有死锁、活锁和饥饿问题。

死锁

多个线程因为环形的等待锁的关系而永远的阻塞下去。

活锁

线程不断重复执行相同的操作，而且总会失败。当多个相互协作的线程都对彼此进行响应而修改各自的状态，并使得任何一个线程都无法继续执行（只能一直重复着响应和修改自身状态），就发生了活锁。如果迎面两个人走路互相让路，总是没有随机性地让到同一个方向，那么就会永远地避让下去。

饥饿

当线程无法访问它所需要的资源而导致无法继续时，就发生了饥饿。如一个线程占有锁永远不释放，等待同一个锁的其他线程就会发生饥饿。

死锁

Java对待死锁的解决方法只要重启程序，所以避免产生死锁十分重要。
死锁主要是因为内嵌锁获取的情况，即占有一个锁并试图获取另一个锁，就很容易和其他的线程发生冲突。

原因

多个线程直接对锁等待的关系产生了环路，如A持有锁A，但是想获得锁B，刚好线程B持有锁B，想要获取锁A，两个线程互相等对方释放锁，就形成了死锁。

简单示例

public class DeathLock {
    private final Object lockA = new Object();
    private final Object lockB = new Object();

    void methodA() throws InterruptedException {
        System.out.println("step in methodA");
        synchronized (lockA) {
            System.out.println("methodA getLockA");
            TimeUnit.SECONDS.sleep(1);
            synchronized (lockB) {
                System.out.println("methodA getLockB");
            }
        }
    }

    void methodB() throws InterruptedException {
        System.out.println("step in methodB");
        synchronized (lockB) {
            TimeUnit.SECONDS.sleep(1);
            System.out.println("methodB getLockB");
            synchronized (lockA) {
                System.out.println("methodB getLockA");
            }
        }
    }

    public static void main(String[] args) {
        DeathLock de = new DeathLock();
        new Thread(() -> {
            try {
                de.methodA();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }).start();
        new Thread(() -> {
            try {
                de.methodB();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }).start();

    }
}

以上示例是很常见的锁顺序死锁，线程A获取锁A之后等待获取锁B，线程B在线程A获取锁A之后获取锁B等待锁A，于是就产生死锁。
上面的死锁的最简单的解决是在两个方法加上synchronized，让他们竞争同一把锁。

使用API检测死锁

ThreadMXBean mbean = ManagementFactory.getThreadMXBean();
        Runnable dlcheck = () -> {
            long[] threads = mbean.findDeadlockedThreads();
            if (threads != null) {
                ThreadInfo[] info = mbean.getThreadInfo(threads);
                System.out.println("detected dead threads:");
                for (ThreadInfo i : info) {
                    System.out.println(i.getThreadName());
                }
            }
        };
        ScheduledExecutorService schedule = Executors.newScheduledThreadPool(1);
        schedule.scheduleAtFixedRate(dlcheck, 5L, 10L, TimeUnit.SECONDS);

示例2 协作对象锁顺序

class Taxi{
    private final Dispatcher dispatcher;
    private Point location;
    
    public Taxi(Dispatcher dispatcher){
        this.dispatcher = dispatcher;
    }
    
    public Point getLocation(){
        return this.location;
    }
    public synchronized void setLocation(Point location){
        this.location = location;
        dispatcher.notifyAvaliable(this);
    }
}

class Dispatcher{
    private final Set<Taxi> taxis;
    private final Set<Taxi> avaliableTaxis;
    public Dispacther(){
        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 taxi : taxis){
            image.dawn(t.getLocation());
        }
        return image;
    }
}

考虑一个线程执行setLocation方法，这个方法先获取Taxi的锁，随后会尝试获取dispatcher的锁，另一个线程在执行getImage方法，这个方法先后需要获取dispatcher的锁和taxi的锁，这两个线程可能会出现死锁问题。要修改可以把内嵌锁拆到外层，使其先释放一个锁再获取另一个锁。
这是隐式的协作产生的死锁，不容易看出来。

示例3

public void switch(Object a,Object b){
    synchronized(a){
        //dosomething
        synchronized(b){
            // dootherthing
        }
    }
}

考虑一个线程执行switch(a,b)，另外一个线程执行switch(b,a)的情况，会发生死锁。

这种可以通过控制锁顺序来解决：

int ahash = system.identityHashCode(a);
int bhash = System.identityHashCode(b);
public void switch(Object a,Object b){
    if( ahash > bhash){
        synchronized(a){
         //dosomething
            synchronized(b){
            // dootherthing
            }
        }
    }else if( ahash < bhash){
        synchronized(b){
         //dosomething
            synchronized(a){
            // dootherthing
            }
        }
    }else{
        sysnchronized(this){
            synchronized(a){
                //dosomething
                synchronized(b){
                    // dootherthing
                }
            }
        }
    }
}

死锁的避免和分析

每次只获取一个锁，避免锁顺序死锁
使用支持定时的锁，可以从死锁中恢复过来。这个机制需要显式锁，内置锁没有这个功能
使用线程转储信息分析死锁

参考资料

[1] Java并发编程实战

03.活跃性问题