Java并发编程学习十二：死锁问题

一、死锁的产生

想要产生死锁，必须同时满足4个必要条件：

• 互斥条件：每个资源每次只能被一个线程（或进程，下同）使用。否则，如果每个人都可以拿到想要的资源，那就不需要等待，所以是不可能发生死锁的
• 请求保持条件：当一个线程因请求资源而阻塞时，则需对已获得的资源保持不放。否则，在请求资源时阻塞后自动释放手中资源（例如锁）的话，其他线程获取到其他的资源，就不会形成死锁
• 不剥夺条件：线程已获得的资源，在未使用完之前，不会被强行剥夺。否则，其他线程可以强行获取资源，同样不会形成死锁
• 循环等待条件：若干线程之间形成一种头尾相接的循环等待资源关系时，才有可能形成死锁。如两个线程互相持有对方所需要的资源，三个及以上的线程形成环路，依次请求下一个线程已经持有的资源

下面是一个必然会发生死锁的案例：

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

    public int flag;
    static Object o1 = new Object();
    static Object o2 = new Object();

    public void run() {
        System.out.println("线程"+Thread.currentThread().getName() + "的flag为" + flag);
        if (flag == 1) {
            synchronized (o1) {
                try {
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (o2) {
                    System.out.println("线程1获得了两把锁");
                }
            }
        }
        if (flag == 2) {
            synchronized (o2) {
                try {
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (o1) {
                    System.out.println("线程2获得了两把锁");
                }
            }
        }
    }

    public static void main(String[] argv) {
        MustDeadLock r1 = new MustDeadLock();
        MustDeadLock r2 = new MustDeadLock();
        r1.flag = 1;
        r2.flag = 2;
        Thread t1 = new Thread(r1, "t1");
        Thread t2 = new Thread(r2, "t2");
        t1.start();
        t2.start();
    }
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

如果 flag 等于 1，就会先获取 o1 这把锁，然后休眠 500 毫秒，再去尝试获取 o2 这把锁并且打印出"线程1获得了两把锁"。如果 flag 等于 2，那么情况恰恰相反，线程会先获取 o2 这把锁，然后休眠 500 毫秒，再去获取 o1 这把锁，并且打印出"线程2获得了两把锁"。

main方法启动两个线程，传入不同的flag，导致两个线程互相持有对象所需要的锁对象，形成死锁。

对应上述四个条件：

• 互斥条件： synchronized 互斥锁保证锁对象 o1、o2 只能同时被一个线程所获得
• 请求保持条件：线程 1 在获得 o1 这把锁之后想去尝试获取 o2 这把锁 ，这时它被阻塞了，但是它并不会自动去释放 o1 这把锁，而是对已获得的资源保持不放。线程2同理。
• 不剥夺条件：JVM 不会主动把某一个线程所持有的锁剥夺
• 循环等待条件：两个线程都想获取对方已持有的资源，形成循环

二、死锁的定位

1. jstack命令

JVM提供了jstack命令，可以用于分析线程状态，发现锁的相互依赖关系。

以上面的MustDeadLock 类为例，程序运行后，在终端执行${JAVA_HOME}/bin/jps命令，可以查看到当前的Java程序的pid

56402 MustDeadLock
56403 Launcher
56474 Jps
55051 KotlinCompileDaemon
1
2
3
4

MustDeadLock对应的pid为56402，继续执行${JAVA_HOME}/bin/jstack 56402命令，打印线程信息

Found one Java-level deadlock:
=============================
"t2":
  waiting to lock monitor 0x00007fa06c004a18 (object 0x000000076adabaf0, a java.lang.Object),
  which is held by "t1"
"t1":
  waiting to lock monitor 0x00007fa06c007358 (object 0x000000076adabb00, a java.lang.Object),
  which is held by "t2"

Java stack information for the threads listed above:
===================================================
"t2":
 at lesson67.MustDeadLock.run(MustDeadLock.java:31)
 - waiting to lock <0x000000076adabaf0> (a java.lang.Object)
 - locked <0x000000076adabb00> (a java.lang.Object)
 at java.lang.Thread.run(Thread.java:748)
"t1":
 at lesson67.MustDeadLock.run(MustDeadLock.java:19)
 - waiting to lock <0x000000076adabb00> (a java.lang.Object)
 - locked <0x000000076adabaf0> (a java.lang.Object)
 at java.lang.Thread.run(Thread.java:748)

Found 1 deadlock
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

打印处的线程信息会提供哪个线程、想要获取哪个锁、形成什么样的环路等信息，还会提供死锁发生的准确位置，从而方便排查。

2. ThreadMXBean工具类

对MustDeadLock 进行改动

public class DetectDeadLock implements Runnable {

    public int flag;
    static Object o1 = new Object();
    static Object o2 = new Object();


    public void run() {
        System.out.println(Thread.currentThread().getName()+" flag = " + flag);
        if (flag == 1) {
            synchronized (o1) {
                try {
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (o2) {
                    System.out.println("线程1获得了两把锁");
                }
            }
        }
        if (flag == 2) {
            synchronized (o2) {
                try {
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (o1) {
                    System.out.println("线程2获得了两把锁");
                }
            }
        }
    }

    public static void main(String[] argv) throws InterruptedException {
        DetectDeadLock r1 = new DetectDeadLock();
        DetectDeadLock r2 = new DetectDeadLock();
        r1.flag = 1;
        r2.flag = 2;
        Thread t1 = new Thread(r1,"t1");
        Thread t2 = new Thread(r2,"t2");
        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("线程id为"+threadInfo.getThreadId()+",线程名为" + threadInfo.getThreadName()+"的线程已经发生死锁，需要的锁正被线程"+threadInfo.getLockOwnerName()+"持有。");
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

通过ThreadMXBean的 findDeadlockedThreads 方法，可以获取到一个 deadlockedThreads 的数组。当这个数组不为空且长度大于 0 的时候，获取对应的线程信息，打印出线程 id，也打印出线程名，同时打印出它所需要的那把锁正被哪个线程所持有。

t1 flag = 1
t2 flag = 2
线程 id 为 12，线程名为 t2 的线程已经发生死锁，需要的锁正被线程 t1 持有。
线程 id 为 11，线程名为 t1 的线程已经发生死锁，需要的锁正被线程 t2 持有。
1
2
3
4

在业务代码中加入ThreadMXBean检测，就可以在发生死锁的时候及时地定位，同时进行报警等其他处理，也就增强了程序的健壮性。

二、死锁的修复

修复死锁的最好时机在于“防患于未然”，而不是事后补救。一旦线上发生死锁问题，为了尽快减小损失，最好的办法是保存 JVM 信息、日志等“案发现场”的数据，然后立刻重启服务，来尝试修复死锁。发生死锁往往要有很多前提条件的，并且当并发度足够高的时候才有可能会发生死锁，所以重启后再次立刻发生死锁的几率并不是很大。重启服务器之后，就可以暂时保证线上服务的可用，然后利用刚才保存过的案发现场的信息，排查死锁、修改代码，最终重新发布。

如何修复死锁问题？其实答案就在死锁产生的四个条件上。由于死锁的产生必须同时满足以上四个条件，因此破坏其中任意一个，就能修复死锁。

下面介绍几种死锁的修复方案

1. 避免策略

发生死锁的一个主要原因是顺序相反的去获取不同的锁，因此可以调整锁的获取顺序来避免死锁，这里相当于破坏循环等待条件。

以银行转账为例，在转账前为了保证线程安全，需要首先获取到两个锁对象，分别是被转出的账户和被转入的账户，但这期间也隐藏着发生死锁的可能性。

public class TransferMoney implements Runnable {

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

    static class Account {

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

        int balance;
    }

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

    public static void transferMoney(Account from, Account to, int amount) {
        //先获取两把锁，然后开始转账
        synchronized (to) {
        	try {
            	Thread.sleep(500);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        }
            synchronized (from) {
                if (from.balance - amount < 0) {
                    System.out.println("余额不足，转账失败。");
                    return;
                }
                from.balance -= amount;
                to.balance += amount;
                System.out.println("成功转账" + amount + "元");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        TransferMoney r1 = new TransferMoney();
        TransferMoney r2 = new TransferMoney();
        r1.flag = 1;
        r2.flag = 0;
        Thread t1 = new Thread(r1);
        Thread t2 = new Thread(r2);
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println("a的余额" + a.balance);
        System.out.println("b的余额" + b.balance);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

定义 int 类型的 flag，它是一个标记位，用于控制不同线程执行不同逻辑。然后建了两个 Account 对象 a 和 b，代表账户，它们最初都有 500 元的余额。

run 方法会根据 flag 值，来决定传入 transferMoney 方法的参数的顺序，如果 flag 为 1，那么就代表从 a 账户转给 b 账户 200元；相反，如果 flag 为 0，那么它就从 b 账户转给 a 账户 200 元。

main方法新建两个 TransferMoney 对象，并且把它们的 flag 分别设置为 1 和 0，然后分别传入两个线程中，并把它们都启动起来。

为了提高死锁发生的可能性，在两个 synchronized 之间加上一个 Thread.sleep(500)，来模拟银行网络迟延等情况。

以上程序会有很大的概率发生死锁，从而导致控制台中不打印任何语句，而且程序也不会停止。最主要原因就是，两个不同的线程获取两个锁的顺序是相反的（第一个线程获取的这两个账户和第二个线程获取的这两个账户顺序恰好相反，第一个线程的“转出账户”正是第二个线程的“转入账户”）

实际上在转账时，并不在意获取两把锁的顺序，只要最终能拿到两把锁，就能进行安全的操作。为了避免两个线程获取锁的顺序相反，这里使用 HashCode 的值来决定顺序，从而保证线程安全。

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

    int fromHash = System.identityHashCode(from);
    int toHash = System.identityHashCode(to);
    if (fromHash < toHash) {
        synchronized (from) {
            synchronized (to) {
                if (from.balance - amount < 0) {
                    System.out.println("余额不足，转账失败。");
                    return;
                }
                from.balance -= amount;
                to.balance += amount;
                System.out.println("成功转账" + amount + "元");
            }
        }
    } else if (fromHash > toHash) {
        synchronized (to) {
            synchronized (from) {
                if (from.balance - amount < 0) {
                    System.out.println("余额不足，转账失败。");
                    return;
                }
                from.balance -= amount;
                to.balance += amount;
                System.out.println("成功转账" + amount + "元");
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

根据 HashCode 的大小来决定获取锁的顺序，不论是哪个线程先执行，不论是转出还是被转入，它获取锁的顺序都会严格根据 HashCode 的值来决定，那么大家获取锁的顺序就一样了，就不会出现获取锁顺序相反的情况。

但是依然有极小的概率会发生 HashCode 相同的情况。实际业务中，一个实体类一般都会有一个主键 ID，主键 ID 具有唯一、不重复的特点，此时使用它的主键 ID 来进行排序，由主键 ID 大小来决定获取锁的顺序，就可以确保避免死锁。

2. 检测与恢复策略

这种策略先允许系统发生死锁，然后再解除。系统可以在每次调用锁的时候，都记录下来调用信息，形成一个“锁的调用链路图”，然后隔一段时间就用死锁检测算法来检测一下，搜索这个图中是否存在环路，一旦发生死锁，就可以用死锁恢复机制，比如剥夺某一个资源，来解开死锁，进行恢复。

这种策略相当于破坏请求保持条件和不剥夺条件。

那么当检测到死锁后，如何解开死锁呢？

a. 终止线程

系统会逐个去终止已经陷入死锁的线程，线程被终止，同时释放资源，这样死锁就会被解开

但是这种终止是要讲究顺序的，需要考虑以下指标：

• 优先级：终止时会考虑到线程或者进程的优先级，先终止优先级低的线程
• 已占用资源和还需要的资源：如果某线程已经占有了一大堆资源，只需要最后一点点资源就可以顺利完成任务，那么系统可能就不会优先选择终止这样的线程，会选择终止别的线程来优先促成该线程的完成
• 已运行时间：当前这个线程已经运行了很多个小时，甚至很多天了，很快就能完成任务了，就不会终止该线程。可以让那些刚刚开始运行的线程终止，并在之后把它们重新启动起来

b. 资源抢占

终止线程的做法有些过于粗暴，其实只需要要把它已经获得的资源进行剥夺，比如让线程回退几步、 释放资源，这样一来就不用终止掉整个线程了。

但这种方式有个缺点，如果抢占的那个线程一直是同一个线程，就会造成线程饥饿。也就是说，这个线程一直被剥夺它已经得到的资源，那么它就长期得不到运行。

3. 鸵鸟策略

鸵鸟在遇到危险时会将头埋到沙子里，这样就看不到危险了。

这种策略的方法就是，认为系统发生死锁的概率不高，并且一旦发生其后果不是特别严重的话，可以选择先忽略它。当死锁发生时候，再人工修复，比如重启服务。

这种策略适用于系统使用的人比较少的情况，如内部的系统，在并发量极低的情况下，它可能几年都不会发生死锁。