Java并发编程 死锁与修复死锁

1.死锁是什么？有什么危害？

1.1 什么是死锁

• 发生在并发中
• 互不相让：当两个（或更多）线程（或进程）相互持有对方所需要的资源，又不主动释放，导致所有人都无法继续前进，导致程序陷入无尽的阻塞，这就是死锁。

image.png

• 多个线程造成死锁的情况

image.png

1.2 死锁的影响

死锁的影响在不同系统中是不一样的，这取决于系统对死锁的处理能力

• 数据库中：检测并放弃事务
• JVM中：无法自动处理

几率不高但危害大

• 不一定发生，但是遵守"墨菲定律"
• 一旦发生，多是高并发场景，影响用户多
• 整个系统崩溃、子系统崩溃、性能降低
• 压力测试无法找出所有潜在的死锁

1.3 发生死锁的例子

必定发生死锁的情况

/**
 * 描述：     必定发生死锁的情况
 */
public class MustDeadLock implements Runnable {

    int flag = 1;

    static Object o1 = new Object();
    static Object o2 = new Object();

    public static void main(String[] args) {
        MustDeadLock r1 = new MustDeadLock();
        MustDeadLock r2 = new MustDeadLock();
        r1.flag = 1;
        r2.flag = 0;
        Thread t1 = new Thread(r1);
        Thread t2 = new Thread(r2);
        t1.start();
        t2.start();
    }

    @Override
    public void run() {
        System.out.println("flag = " + flag);
        if (flag == 1) {
            synchronized (o1) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (o2) {
                    System.out.println("线程1成功拿到两把锁");
                }
            }
        }
        if (flag == 0) {
            synchronized (o2) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (o1) {
                    System.out.println("线程2成功拿到两把锁");
                }
            }
        }
    }
}

flag = 1
flag = 0

多个人相互转账

• 需要两把锁
• 获取两把锁成功，且余额大于0，则扣除转出人，增加收款人的余额，是原子操作
• 顺序相反导致死锁

public class MultiTransferMoney {

    private static final int NUM_ACCOUNTS = 500;
    private static final int NUM_MONEY = 1000;
    private static final int NUM_ITERATIONS = 1000000;
    private static final int NUM_THREADS = 100;

    public static void main(String[] args) {

        Random rnd = new Random();
        TransferMoney.Account[] accounts = new TransferMoney.Account[NUM_ACCOUNTS];
        for (int i = 0; i < accounts.length; i++) {
            accounts[i] = new TransferMoney.Account(NUM_MONEY);
        }
        class TransferThread extends Thread {

            @Override
            public void run() {
                for (int i = 0; i < NUM_ITERATIONS; i++) {
                    int fromAcct = rnd.nextInt(NUM_ACCOUNTS);
                    int toAcct = rnd.nextInt(NUM_ACCOUNTS);
                    int amount = rnd.nextInt(NUM_MONEY);
                    TransferMoney.transferMoney(accounts[fromAcct], accounts[toAcct], amount);
                }
                System.out.println("运行结束");
            }
        }
        for (int i = 0; i < NUM_THREADS; i++) {
            new TransferThread().start();
        }
    }
}

public class TransferMoney implements Runnable {

    static Account a = new Account(500);
    static Account b = new Account(500);
    static Object lock = new Object();
    int flag = 1;

    public static void transferMoney(Account from, Account to, int amount) {
        class Helper {

            public void transfer() {
                if (from.balance - amount < 0) {
                    System.out.println("余额不足，转账失败。");
                    return;
                }
                from.balance -= amount;
                to.balance = to.balance + amount;
                System.out.println("成功转账" + amount + "元");
            }
        }

        synchronized (from) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (to) {
                new Helper().transfer();
            }
        }
    }

    @Override
    public void run() {
        if (flag == 1) {
            transferMoney(a, b, 200);
        }
        if (flag == 0) {
            transferMoney(b, a, 200);
        }
    }

    static class Account {

        int balance;

        public Account(int balance) {
            this.balance = balance;
        }
    }
}

死锁

image.png

程序停止输出发生死锁

2. 死锁的4个必要条件:

1. 互斥条件
2. 请求与保持条件
3. 不剥夺条件
4. 循环等待条件

3. 如何定位死锁

3.1.jstack定位死锁

并发编程中的死锁定位排查

image.png image.png

3.2.ThreadMXBean定位死锁

public class ThreadMXBeanDetection implements Runnable {

    int flag = 1;

    static Object o1 = new Object();
    static Object o2 = new Object();

    public static void main(String[] args) throws InterruptedException {
        ThreadMXBeanDetection r1 = new ThreadMXBeanDetection();
        ThreadMXBeanDetection r2 = new ThreadMXBeanDetection();
        r1.flag = 1;
        r2.flag = 0;
        Thread t1 = new Thread(r1);
        Thread t2 = new Thread(r2);
        t1.start();
        t2.start();
        Thread.sleep(1000);
        ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
        long[] deadlockedThreads = threadMXBean.findDeadlockedThreads();
        if (deadlockedThreads != null && deadlockedThreads.length > 0) {
            for (int i = 0; i < deadlockedThreads.length; i++) {
                ThreadInfo threadInfo = threadMXBean.getThreadInfo(deadlockedThreads[i]);
                System.out.println("发现死锁" + threadInfo.getThreadName());
            }
        }
    }

    @Override
    public void run() {
        System.out.println("flag = " + flag);
        if (flag == 1) {
            synchronized (o1) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (o2) {
                    System.out.println("线程1成功拿到两把锁");
                }
            }
        }
        if (flag == 0) {
            synchronized (o2) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (o1) {
                    System.out.println("线程2成功拿到两把锁");
                }
            }
        }
    }
}

flag = 0
flag = 1
发现死锁Thread-1
发现死锁Thread-0

4. 修复死锁策略

线上发生死锁应该什么办

• 线上问题都需要防患与未然，不造成损失几乎是不可能
• 保存死锁数据，然后立刻重启服务器
• 暂时保证线上服务的安全，然后再利用刚才保存的信息，排查死锁，修改代码，重新发版

常见修复策略

4.1. 避免策略

示例1 修改两人转账时获取锁的顺序

经过思考，我们可以发现，其实转账时，并不在乎两把锁的相对获取顺序。转账的时候，我们无论先获取到转出账户锁对象，还是先获取到转入账户锁对象，只要最终能拿到两把锁，就能进行安全的操作。所以我们来调整一下获取锁的顺序，使得先获取的账户和该账户是“转入”或“转出”无关，而是使用 HashCode 的值来决定顺序，从而保证线程安全

    public static void transferMoney(Account from, Account to, int amount) {
        class Helper {

            public void transfer() {
                if (from.balance - amount < 0) {
                    System.out.println("余额不足，转账失败。");
                    return;
                }
                from.balance -= amount;
                to.balance = to.balance + amount;
                System.out.println("成功转账" + amount + "元");
            }
        }
        int fromHash = System.identityHashCode(from);
        int toHash = System.identityHashCode(to);
        if (fromHash < toHash) {
            synchronized (from) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (to) {
                    new Helper().transfer();
                }
            }
        }
        else if (fromHash > toHash) {
            synchronized (to) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (from) {
                    new Helper().transfer();
                }
            }
        }else  {
            synchronized (lock) {
                synchronized (to) {
                    synchronized (from) {
                        new Helper().transfer();
                    }
                }
            }
        }

    }

可以看到，我们会分别计算出这两个 Account 的 HashCode，然后根据 HashCode 的大小来决定获取锁的顺序。这样一来，不论是哪个线程先执行，不论是转出还是被转入，它获取锁的顺序都会严格根据 HashCode 的值来决定，那么大家获取锁的顺序就一样了，就不会出现获取锁顺序相反的情况，也就避免了死锁

总结：通过hashcode来决定获取锁的顺序、冲突时需要“加时赛”(再加一把锁)获取锁

image.png

示例2 哲学家换手解决.改变一个哲学家拿筷子的顺序.

/**
 * 描述：     演示哲学家就餐问题导致的死锁
 */
public class DiningForPhilosophers {

    /**
     * 哲学家类
     */
    public static class Philosophers implements Runnable{

        private Object leftChopsticks;
        private Object rightChopsticks;

        public Philosophers(Object leftChopsticks, Object rightChopsticks) {
            this.leftChopsticks = leftChopsticks;
            this.rightChopsticks = rightChopsticks;
        }

        @Override
        public void run() {
            try {
                while(true){
                    action("思考......");
                    synchronized (leftChopsticks){
                        action("拿起左边的筷子");
                        synchronized (rightChopsticks){
                            action("拿起右边的筷子---吃饭");
                            action("放下右边的筷子");
                        }
                        action("放下左边的筷子");
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public static void action(String action) throws InterruptedException {
            System.out.println(Thread.currentThread().getName()+"-->"+action);
            //随机休眠,代表这个行为执行的耗时
            Thread.sleep((long) (Math.random()*10));
        }
    };

    public static void main(String[] args) {
        Philosophers[] philosophers = new Philosophers[5];
        Object[] chopsticks = new Object[philosophers.length];
        //初始化筷子对象
        for (int i=0;i<chopsticks.length;i++){
            chopsticks[i] = new Object();
        }
        //初始化哲学家对象,并启动线程.
        for (int i=0;i<chopsticks.length;i++){
            Object leftChopsticks = chopsticks[i];
            Object rightChopsticks = chopsticks[(i+1)%chopsticks.length];
            philosophers[i] = new Philosophers(leftChopsticks, rightChopsticks);
            new Thread(philosophers[i],"哲学家"+i+"号").start();
        }
    }

}

image.png

同样发生死锁,这里大家陷入了一种循环等待的状态,0号获取了他左边的0号筷子,请求他右边的1号筷子,1号获取了左边的1号筷子,请求等待他右边的2号筷子...........5号获取了他左边的5号筷子等待他右手边的0号筷子........这样就形成了一个环.

哲学家问题解决策略

• 服务员检查
  哲学家拿叉子时 服务员检查叉子数量
• 改变一个哲学家拿叉子的顺序
• 餐票
  哲学家就餐时先拿到餐票就能就餐
• 领导调节（检测与恢复策略）

这里演示改变一个哲学家拿叉子的顺序修复

image.png
image.png

这样就成功的避免了死锁

4.2 检测与恢复策略：一段时间检测是否有死锁,如果有就剥夺某个资源,来解除死锁

允许死锁的发生,但是发生死锁后要记录下来并通过停止线程或其他方式停止死锁

死锁检测算法

• 允许发生死锁
• 每次调用锁的记录
• 定期检查"锁的调用链路图"中是否存在环路
• 一旦发生死锁，就用死锁恢复机制进行恢复机制进行恢复

锁的调用链路图

恢复方法1：进程终止

• 逐个终止线程，直到死锁消除。
• 终止顺序：
  1. 优先级（重要性，是前台交互还是后台处理）
  2. 已占用资源、还需要的资源
  3. 已经运行时间

恢复方法2：资源抢占

• 把已经分发出去的锁给收回来
• 让线程回退几步，这样就不用结束整个线程，成本比较低
  比如 让哲学家把拿起的筷子再放下
• 缺点：可能同一个线程一直被抢占，那就造成饥饿

4.3 鸵鸟策略

死锁发生的几率特别小,忽略他，等死锁发生了,再去处理修改

5 实际工程中如何避免死锁

1. 设置超时时间

• Lock的tryLock(long timeout,TimeUnit unit)
  造成超时的可能性很多,发生了死锁,线程陷入了死循环,线程执行很慢.

/**
 * 描述：     用tryLock来避免死锁
 */
public class TryLockDeadlock implements Runnable {

    int flag = 1;
    static Lock lock1 = new ReentrantLock();
    static Lock lock2 = new ReentrantLock();

    public static void main(String[] args) {
        TryLockDeadlock r1 = new TryLockDeadlock();
        TryLockDeadlock r2 = new TryLockDeadlock();
        r1.flag = 1;
        r2.flag = 0;
        new Thread(r1).start();
        new Thread(r2).start();
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            if (flag == 1) {
                try {
                    if (lock1.tryLock(800, TimeUnit.MILLISECONDS)) {
                        System.out.println("线程1获取到了锁1");
                        Thread.sleep(new Random().nextInt(1000));
                        if (lock2.tryLock(800, TimeUnit.MILLISECONDS)) {
                            System.out.println("线程1获取到了锁2");
                            System.out.println("线程1成功获取到了两把锁");
                            lock2.unlock();
                            lock1.unlock();
                            break;
                        } else {
                            System.out.println("线程1尝试获取锁2失败，已重试");
                            lock1.unlock();
                            Thread.sleep(new Random().nextInt(1000));
                        }
                    } else {
                        System.out.println("线程1获取锁1失败，已重试");
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            if (flag == 0) {
                try {
                    if (lock2.tryLock(3000, TimeUnit.MILLISECONDS)) {
                        System.out.println("线程2获取到了锁2");

                        Thread.sleep(new Random().nextInt(1000));
                        if (lock1.tryLock(3000, TimeUnit.MILLISECONDS)) {
                            System.out.println("线程2获取到了锁1");
                            System.out.println("线程2成功获取到了两把锁");
                            lock1.unlock();
                            lock2.unlock();
                            break;
                        } else {
                            System.out.println("线程2尝试获取锁1失败，已重试");
                            lock2.unlock();
                            Thread.sleep(new Random().nextInt(1000));
                        }
                    } else {
                        System.out.println("线程2获取锁2失败，已重试");
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

线程1获取到了锁1
线程2获取到了锁2
线程1尝试获取锁2失败，已重试
线程2获取到了锁1
线程2成功获取到了两把锁
线程1获取到了锁1
线程1获取到了锁2
线程1成功获取到了两把锁

2.多使用并发类而不是自己设计锁

如ConcurrentHashMap
java.util.concurrent.atomic.

3.尽量降低锁的使用粒度:用不同的锁而不是一个锁

缩小锁的临界区

4.如果能使用同步代码块,就不使用同步方法:自己指定锁对象

缩小了同步范围,可以自己指定锁对象

5.给线程指定有意义的名字,方便后期debug和排查

6.避免锁的嵌套:MustDeadLock 演示的嵌套

7.分配资源前先看能不能收回来：银行家算法

银行家算法（Java实现）

8.尽量不要几个功能用通一把锁：专锁专用

特别感谢：

悟空