04共享模型之管程（下）：wait/notify与ReentrantLock

本文是管程部分的下篇，聚焦线程间的协作：为什么有了 synchronized 还需要 wait/notify，wait/notify 的正确使用姿势，Park/Unpark 的精确唤醒，六种线程状态之间的转换规则，多把锁带来的并发度提升与死锁等活跃性问题，最后讲解功能更强大的 ReentrantLock。

4.7 wait notify

小故事：为什么需要 wait

由于条件不满足，小南不能继续进行计算。
但小南如果一直占着锁，其它人就得一直阻塞，效率太低。
于是老王单开了一间休息室（调用 wait 方法），让小南到休息室（WaitSet）等着，这时锁被释放，其它人可以由老王随机安排进屋。
直到小 M 将烟送来，大叫一声「你的烟到了」（调用 notify 方法）。
小南于是离开休息室，重新进入竞争锁的队列。

条件不满足时，Owner 线程占着锁无法继续干活

调用 wait 后线程进入 WaitSet 释放锁，notify 后线程被唤醒

被唤醒的线程离开 WaitSet，进入 EntryList 重新竞争锁

API 介绍

obj.wait()：让进入 object 监视器的线程到 WaitSet 等待；
obj.notify()：在 object 上正在 WaitSet 等待的线程中挑一个唤醒；
obj.notifyAll()：让 object 上正在 WaitSet 等待的线程全部唤醒。

它们都是线程之间进行协作的手段，属于 Object 对象的方法。必须先获得此对象的锁，才能调用这几个方法。

基本使用

final static Object obj = new Object();
 
public static void main(String[] args) {
 
    new Thread(() -> {
        synchronized (obj) {
            log.debug("执行....");
            try {
                obj.wait(); // 让线程在obj上一直等待下去
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("其它代码....");
        }
    }).start();
 
    new Thread(() -> {
        synchronized (obj) {
            log.debug("执行....");
            try {
                obj.wait(); // 让线程在obj上一直等待下去
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("其它代码....");
        }
    }).start();
 
    // 主线程两秒后执行
    sleep(2);
    log.debug("唤醒 obj 上其它线程");
    synchronized (obj) {
        obj.notify(); // 唤醒obj上一个线程
        // obj.notifyAll(); // 唤醒obj上所有等待线程
    }
}

使用 notify 的一种结果：

20:00:53.096 [Thread-0] c.TestWaitNotify - 执行....
20:00:53.099 [Thread-1] c.TestWaitNotify - 执行....
20:00:55.096 [main] c.TestWaitNotify - 唤醒 obj 上其它线程
20:00:55.096 [Thread-0] c.TestWaitNotify - 其它代码....

使用 notifyAll 的结果：

19:58:15.457 [Thread-0] c.TestWaitNotify - 执行....
19:58:15.460 [Thread-1] c.TestWaitNotify - 执行....
19:58:17.456 [main] c.TestWaitNotify - 唤醒 obj 上其它线程
19:58:17.456 [Thread-1] c.TestWaitNotify - 其它代码....
19:58:17.456 [Thread-0] c.TestWaitNotify - 其它代码....

wait() 会释放对象的锁，进入 WaitSet 等待区，从而让其他线程有机会获取对象的锁。无限制等待，直到被 notify 为止。
wait(long n)：有时限的等待，到 n 毫秒后结束等待，或是被 notify 提前唤醒。

4.8 wait notify 的正确姿势

sleep(long n) 和 wait(long n) 的区别

sleep 是 Thread 的方法，而 wait 是 Object 的方法；
sleep 不需要强制和 synchronized 配合使用，但 wait 必须和 synchronized 一起用；
sleep 在睡眠时不会释放对象锁，wait 在等待时会释放对象锁；
两者的等待状态都是 TIMED_WAITING（wait 不带超时为 WAITING）。

step 1：用 sleep 等待条件，有什么问题？

static final Object room = new Object();
static boolean hasCigarette = false;
static boolean hasTakeout = false;
 
new Thread(() -> {
    synchronized (room) {
        log.debug("有烟没？[{}]", hasCigarette);
        if (!hasCigarette) {
            log.debug("没烟，先歇会！");
            sleep(2);
        }
        log.debug("有烟没？[{}]", hasCigarette);
        if (hasCigarette) {
            log.debug("可以开始干活了");
        }
    }
}, "小南").start();
 
for (int i = 0; i < 5; i++) {
    new Thread(() -> {
        synchronized (room) {
            log.debug("可以开始干活了");
        }
    }, "其它人").start();
}
 
sleep(1);
new Thread(() -> {
    // 这里能不能加 synchronized (room)？
    hasCigarette = true;
    log.debug("烟到了噢！");
}, "送烟的").start();

输出：

20:49:49.883 [小南] c.TestCorrectPosture - 有烟没？[false]
20:49:49.887 [小南] c.TestCorrectPosture - 没烟，先歇会！
20:49:50.882 [送烟的] c.TestCorrectPosture - 烟到了噢！
20:49:51.887 [小南] c.TestCorrectPosture - 有烟没？[true]
20:49:51.887 [小南] c.TestCorrectPosture - 可以开始干活了
20:49:51.887 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.887 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.888 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.888 [其它人] c.TestCorrectPosture - 可以开始干活了
20:49:51.888 [其它人] c.TestCorrectPosture - 可以开始干活了

问题：

其它干活的线程都要一直阻塞，效率太低；
小南必须睡足 2s 才能醒来，即使烟提前送到也无法立刻醒来；
加了 synchronized (room) 后，就好比小南在里面反锁门睡觉，烟根本送不进门；main 没加 synchronized 就好像是翻窗户进来修改共享变量。

解决方法：使用 wait - notify 机制。

step 2：改用 wait(2000)

new Thread(() -> {
    synchronized (room) {
        log.debug("有烟没？[{}]", hasCigarette);
        if (!hasCigarette) {
            log.debug("没烟，先歇会！");
            try {
                room.wait(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        log.debug("有烟没？[{}]", hasCigarette);
        if (hasCigarette) {
            log.debug("可以开始干活了");
        }
    }
}, "小南").start();
 
for (int i = 0; i < 5; i++) {
    new Thread(() -> {
        synchronized (room) {
            log.debug("可以开始干活了");
        }
    }, "其它人").start();
}
 
sleep(1);
new Thread(() -> {
    synchronized (room) {
        hasCigarette = true;
        log.debug("烟到了噢！");
        room.notify();
    }
}, "送烟的").start();

输出：

20:51:42.489 [小南] c.TestCorrectPosture - 有烟没？[false]
20:51:42.493 [小南] c.TestCorrectPosture - 没烟，先歇会！
20:51:42.493 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.493 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.494 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.494 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:42.494 [其它人] c.TestCorrectPosture - 可以开始干活了
20:51:43.490 [送烟的] c.TestCorrectPosture - 烟到了噢！
20:51:43.490 [小南] c.TestCorrectPosture - 有烟没？[true]
20:51:43.490 [小南] c.TestCorrectPosture - 可以开始干活了

这解决了其它干活线程被阻塞的问题。但如果有其它线程也在等待不同条件呢？

step 3：notify 可能造成虚假唤醒

增加一个等外卖的小女：

new Thread(() -> {
    synchronized (room) {
        log.debug("有烟没？[{}]", hasCigarette);
        if (!hasCigarette) {
            log.debug("没烟，先歇会！");
            try {
                room.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        log.debug("有烟没？[{}]", hasCigarette);
        if (hasCigarette) {
            log.debug("可以开始干活了");
        } else {
            log.debug("没干成活...");
        }
    }
}, "小南").start();
 
new Thread(() -> {
    synchronized (room) {
        Thread thread = Thread.currentThread();
        log.debug("外卖送到没？[{}]", hasTakeout);
        if (!hasTakeout) {
            log.debug("没外卖，先歇会！");
            try {
                room.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        log.debug("外卖送到没？[{}]", hasTakeout);
        if (hasTakeout) {
            log.debug("可以开始干活了");
        } else {
            log.debug("没干成活...");
        }
    }
}, "小女").start();
 
sleep(1);
new Thread(() -> {
    synchronized (room) {
        hasTakeout = true;
        log.debug("外卖到了噢！");
        room.notify();
    }
}, "送外卖的").start();

输出：

20:53:12.173 [小南] c.TestCorrectPosture - 有烟没？[false]
20:53:12.176 [小南] c.TestCorrectPosture - 没烟，先歇会！
20:53:12.176 [小女] c.TestCorrectPosture - 外卖送到没？[false]
20:53:12.176 [小女] c.TestCorrectPosture - 没外卖，先歇会！
20:53:13.174 [送外卖的] c.TestCorrectPosture - 外卖到了噢！
20:53:13.174 [小南] c.TestCorrectPosture - 有烟没？[false]
20:53:13.174 [小南] c.TestCorrectPosture - 没干成活...

notify 只能随机唤醒一个 WaitSet 中的线程，这时可能唤醒不了正确的线程（外卖到了却唤醒了等烟的小南），称之为【虚假唤醒】。解决方法：改为 notifyAll。

step 4：改用 notifyAll

new Thread(() -> {
    synchronized (room) {
        hasTakeout = true;
        log.debug("外卖到了噢！");
        room.notifyAll();
    }
}, "送外卖的").start();

输出：

20:55:23.978 [小南] c.TestCorrectPosture - 有烟没？[false]
20:55:23.982 [小南] c.TestCorrectPosture - 没烟，先歇会！
20:55:23.982 [小女] c.TestCorrectPosture - 外卖送到没？[false]
20:55:23.982 [小女] c.TestCorrectPosture - 没外卖，先歇会！
20:55:24.979 [送外卖的] c.TestCorrectPosture - 外卖到了噢！
20:55:24.979 [小女] c.TestCorrectPosture - 外卖送到没？[true]
20:55:24.980 [小女] c.TestCorrectPosture - 可以开始干活了
20:55:24.980 [小南] c.TestCorrectPosture - 有烟没？[false]
20:55:24.980 [小南] c.TestCorrectPosture - 没干成活...

notifyAll 解决了所有线程都被唤醒的问题，但 if + wait 只判断一次条件：小南醒来后条件仍然不成立，却没有重新判断的机会了。解决方法：用 while + wait，条件不成立就再次 wait。

step 5：将 if 改为 while

while (!hasCigarette) {
    log.debug("没烟，先歇会！");
    try {
        room.wait();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

输出：

20:58:34.322 [小南] c.TestCorrectPosture - 有烟没？[false]
20:58:34.326 [小南] c.TestCorrectPosture - 没烟，先歇会！
20:58:34.326 [小女] c.TestCorrectPosture - 外卖送到没？[false]
20:58:34.326 [小女] c.TestCorrectPosture - 没外卖，先歇会！
20:58:35.323 [送外卖的] c.TestCorrectPosture - 外卖到了噢！
20:58:35.324 [小女] c.TestCorrectPosture - 外卖送到没？[true]
20:58:35.324 [小女] c.TestCorrectPosture - 可以开始干活了
20:58:35.324 [小南] c.TestCorrectPosture - 没烟，先歇会！

wait/notify 的正确使用模板

推荐套路：等待方永远在 while 循环中检查条件，唤醒方使用 notifyAll：

synchronized(lock) {
    while(条件不成立) {
        lock.wait();
    }
    // 干活
}
 
//另一个线程
synchronized(lock) {
    lock.notifyAll();
}

为什么等待必须用 while？因为被唤醒后到重新获得锁之间，条件可能已经被别的线程改变（虚假唤醒/竞态），必须再次检查确认条件成立才能继续干活。

模式：保护性暂停与生产者消费者

**保护性暂停（Guarded Suspension）**是一种同步模式：一个线程等待另一个线程的执行结果，结果没到就一直等待，结果到了再继续。其核心就是上面的 while + wait/notifyAll 模板，等待方和结果提供方在同一个管程上同步汇合。
生产者/消费者是一种异步模式：生产者和消费者之间通过消息队列解耦，生产者不必等待消费者处理完，二者各自异步运行（通常也用 wait/notify 实现队列的等待与唤醒）。

4.9 Park & Unpark

基本使用

Park 与 Unpark 是 LockSupport 类中的方法：

// 暂停当前线程
LockSupport.park();
 
// 恢复某个线程的运行
LockSupport.unpark(暂停线程对象)

先 park 再 unpark：

Thread t1 = new Thread(() -> {
    log.debug("start...");
    sleep(1);
    log.debug("park...");
    LockSupport.park();
    log.debug("resume...");
},"t1");
t1.start();
 
sleep(2);
log.debug("unpark...");
LockSupport.unpark(t1);

输出：

18:42:52.585 c.TestParkUnpark [t1] - start...
18:42:53.589 c.TestParkUnpark [t1] - park...
18:42:54.583 [main] c.TestParkUnpark - unpark...
18:42:54.583 [TestParkUnpark [t1] - resume...

先 unpark 再 park：

Thread t1 = new Thread(() -> {
    log.debug("start...");
    sleep(2);
    log.debug("park...");
    LockSupport.park();
    log.debug("resume...");
}, "t1");
t1.start();
 
sleep(1);
log.debug("unpark...");
LockSupport.unpark(t1);

输出：

18:43:50.765 c.TestParkUnpark [t1] - start...
18:43:51.764 c.TestParkUnpark [main] - unpark...
18:43:52.769 c.TestParkUnpark [t1] - park...
18:43:52.769 c.TestParkUnpark [t1] - resume...

提前 unpark 发放了「许可」，之后 park 不会暂停，直接继续运行。

特点

与 Object 的 wait / notify 相比：

wait、notify 和 notifyAll 必须配合 Object Monitor 一起使用，而 park、unpark 不必在 synchronized 块中使用；
park / unpark 是以线程为单位来【阻塞】和【唤醒】的，而 notify 只能随机唤醒一个等待线程，notifyAll 唤醒所有等待线程，不那么【精确】；
park / unpark 可以先 unpark，而 wait / notify 不能先 notify。

基本原理

每个线程都有一个 Parker，内部维护一个计数器（许可）：

调用 unpark 时，将该线程的许可置为 1（多次调用也不会累加，最多为 1），并唤醒线程；
调用 park 时，如果许可为 1，就把许可清零并立即返回；如果许可为 0，则阻塞等待。

所以「先 unpark 再 park」也能正确放行。

4.10 重新理解线程状态转换

Java 线程六种状态（NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING、TERMINATED）转换总图

假设有线程 Thread t：

情况 1：NEW --> RUNNABLE

当调用 t.start() 方法时，由 NEW --> RUNNABLE。

情况 2：RUNNABLE <--> WAITING（wait）

t 线程用 synchronized(obj) 获取了对象锁后：

调用 obj.wait() 方法时，t 从 RUNNABLE --> WAITING；
调用 obj.notify()、obj.notifyAll()、t.interrupt() 时：
- 竞争锁成功，t 从 WAITING --> RUNNABLE；
- 竞争锁失败，t 从 WAITING --> BLOCKED。

public class TestWaitNotify {
    final static Object obj = new Object();
 
    public static void main(String[] args) {
 
        new Thread(() -> {
            synchronized (obj) {
                log.debug("执行....");
                try {
                    obj.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("其它代码...."); // 断点
            }
        },"t1").start();
 
        new Thread(() -> {
            synchronized (obj) {
                log.debug("执行....");
                try {
                    obj.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("其它代码...."); // 断点
            }
        },"t2").start();
 
        sleep(0.5);
        log.debug("唤醒 obj 上其它线程");
        synchronized (obj) {
            obj.notifyAll(); // 唤醒obj上所有等待线程  断点
        }
    }
}

情况 3：RUNNABLE <--> WAITING（join）

当前线程调用 t.join() 方法时，当前线程从 RUNNABLE --> WAITING（注意是当前线程在 t 线程对象的监视器上等待）；
t 线程运行结束，或调用了当前线程的 interrupt() 时，当前线程从 WAITING --> RUNNABLE。

情况 4：RUNNABLE <--> WAITING（park）

当前线程调用 LockSupport.park() 会让当前线程从 RUNNABLE --> WAITING；
调用 LockSupport.unpark(目标线程) 或调用了线程的 interrupt()，会让目标线程从 WAITING --> RUNNABLE。

情况 5：RUNNABLE <--> TIMED_WAITING（带超时的 wait）

t 线程用 synchronized(obj) 获取了对象锁后：

调用 obj.wait(long n) 时，t 从 RUNNABLE --> TIMED_WAITING；
等待超过 n 毫秒，或调用 notify、notifyAll、interrupt 时：
- 竞争锁成功，t 从 TIMED_WAITING --> RUNNABLE；
- 竞争锁失败，t 从 TIMED_WAITING --> BLOCKED。

情况 6：RUNNABLE <--> TIMED_WAITING（带超时的 join）

当前线程调用 t.join(long n) 时，当前线程从 RUNNABLE --> TIMED_WAITING（当前线程在 t 线程对象的监视器上等待）；
等待超过 n 毫秒，或 t 线程运行结束，或调用当前线程的 interrupt() 时，从 TIMED_WAITING --> RUNNABLE。

情况 7：RUNNABLE <--> TIMED_WAITING（sleep）

当前线程调用 Thread.sleep(long n)，从 RUNNABLE --> TIMED_WAITING；
等待超过 n 毫秒，从 TIMED_WAITING --> RUNNABLE。

情况 8：RUNNABLE <--> TIMED_WAITING（parkNanos / parkUntil）

当前线程调用 LockSupport.parkNanos(long nanos) 或 LockSupport.parkUntil(long millis) 时，从 RUNNABLE --> TIMED_WAITING；
调用 unpark、interrupt，或是等待超时，会让目标线程从 TIMED_WAITING --> RUNNABLE。

情况 9：RUNNABLE <--> BLOCKED

t 线程用 synchronized(obj) 获取对象锁时如果竞争失败，从 RUNNABLE --> BLOCKED；
持锁线程的同步代码块执行完毕，会唤醒该对象上所有 BLOCKED 的线程重新竞争，其中 t 竞争成功则从 BLOCKED --> RUNNABLE，其它失败的线程仍然 BLOCKED。

情况 10：RUNNABLE <--> TERMINATED

当前线程所有代码运行完毕，进入 TERMINATED。

4.11 多把锁

多把不相干的锁

一间大屋子有睡觉、学习两个互不相干的功能。小南要学习、小女要睡觉，如果只用一间屋子（一个对象锁），并发度很低。解决方法是准备多个房间（多个对象锁）。

例如：

class BigRoom {
 
    public void sleep() {
        synchronized (this) {
            log.debug("sleeping 2 小时");
            Sleeper.sleep(2);
        }
    }
 
    public void study() {
        synchronized (this) {
            log.debug("study 1 小时");
            Sleeper.sleep(1);
        }
    }
 
}
 
BigRoom bigRoom = new BigRoom();
new Thread(() -> {
    bigRoom.compute();
},"小南").start();
new Thread(() -> {
    bigRoom.sleep();
},"小女").start();

某次结果（两人必须串行，总耗时约 3 小时）：

12:13:54.471 [小南] c.BigRoom - study 1 小时
12:13:55.476 [小女] c.BigRoom - sleeping 2 小时

改进：

class BigRoom {
 
    private final Object studyRoom = new Object();
    private final Object bedRoom = new Object();
 
    public void sleep() {
        synchronized (bedRoom) {
            log.debug("sleeping 2 小时");
            Sleeper.sleep(2);
        }
    }
 
    public void study() {
        synchronized (studyRoom) {
            log.debug("study 1 小时");
            Sleeper.sleep(1);
        }
    }
 
}

某次执行结果（两人并行）：

12:15:35.069 [小南] c.BigRoom - study 1 小时
12:15:35.069 [小女] c.BigRoom - sleeping 2 小时

将锁的粒度细分：

好处：可以增强并发度；
坏处：如果一个线程需要同时获得多把锁，就容易发生死锁。

4.12 活跃性

死锁

一个线程需要同时获取多把锁时，就容易发生死锁。t1 线程获得 A 对象锁，接下来想获取 B 对象锁；t2 线程获得 B 对象锁，接下来想获取 A 对象锁：

Object A = new Object();
Object B = new Object();
Thread t1 = new Thread(() -> {
    synchronized (A) {
        log.debug("lock A");
        sleep(1);
        synchronized (B) {
            log.debug("lock B");
            log.debug("操作...");
        }
    }
}, "t1");
 
Thread t2 = new Thread(() -> {
    synchronized (B) {
        log.debug("lock B");
        sleep(0.5);
        synchronized (A) {
            log.debug("lock A");
            log.debug("操作...");
        }
    }
}, "t2");
t1.start();
t2.start();

结果：

12:22:06.962 [t2] c.TestDeadLock - lock B
12:22:06.962 [t1] c.TestDeadLock - lock A

之后程序永远卡住，谁也无法继续执行。

定位死锁

检测死锁可以使用 jconsole 工具，或者使用 jps 定位进程 id，再用 jstack 定位死锁：

cmd > jps
Picked up JAVA_TOOL_OPTIONS: -Dfile.encoding=UTF-8
12320 Jps
22816 KotlinCompileDaemon
33200 TestDeadLock              // JVM 进程
11508 Main
28468 Launcher
 
cmd > jstack 33200
 
"Thread-1" #12 prio=5 os_prio=0 tid=0x000000001eb69000 nid=0xd40 waiting for monitor entry
   java.lang.Thread.State: BLOCKED (on object monitor)
        at thread.TestDeadLock.lambda$main$1(TestDeadLock.java:28)
        - waiting to lock <0x000000076b5bf1c0> (a java.lang.Object)
        - locked <0x000000076b5bf1d0> (a java.lang.Object)
        at java.lang.Thread.run(Thread.java:745)
 
"Thread-0" #11 prio=5 os_prio=0 tid=0x000000001eb68800 nid=0x1b28 waiting for monitor entry
   java.lang.Thread.State: BLOCKED (on object monitor)
        at thread.TestDeadLock.lambda$main$0(TestDeadLock.java:15)
        - waiting to lock <0x000000076b5bf1d0> (a java.lang.Object)
        - locked <0x000000076b5bf1c0> (a java.lang.Object)
        at java.lang.Thread.run(Thread.java:745)
 
Found one Java-level deadlock:
=============================
"Thread-1":
  waiting to lock monitor 0x000000000361d378 (object 0x000000076b5bf1c0, a java.lang.Object),
  which is held by "Thread-0"
"Thread-0":
  waiting to lock monitor 0x000000000361e768 (object 0x000000076b5bf1d0, a java.lang.Object),
  which is held by "Thread-1"
 
Found 1 deadlock.

避免死锁要注意加锁顺序。另外如果某个线程进入了死循环导致其它线程一直等待，Linux 下可以先通过 top 定位 CPU 占用高的 Java 进程，再用 top -Hp 进程id 定位是哪个线程，最后用 jstack 排查。

死锁产生的四个必要条件

产生死锁需要同时满足以下四个条件，破坏其中任意一个即可避免死锁：

互斥条件：资源在同一时刻只能被一个线程占用；
占有且等待：线程已经持有至少一个资源，又去等待被其它线程占有的资源；
不可抢占：资源不能被强行夺走，只能由持有者主动释放；
循环等待：一组线程首尾相接，形成等待环。

固定加锁顺序（所有线程都按相同顺序获取锁）破坏的就是「循环等待」条件。

哲学家就餐问题

有五位哲学家围坐在圆桌旁：

他们只做思考和吃饭两件事，思考一会儿吃口饭，吃完接着思考；
吃饭要用两根筷子，桌上共有 5 根筷子，每位哲学家左右手边各有一根；
如果筷子被身边的人拿着，自己就得等待。

筷子类：

class Chopstick {
    String name;
 
    public Chopstick(String name) {
        this.name = name;
    }
 
    @Override
    public String toString() {
        return "筷子{" + name + '}';
    }
}

哲学家类：

class Philosopher extends Thread {
    Chopstick left;
    Chopstick right;
 
    public Philosopher(String name, Chopstick left, Chopstick right) {
        super(name);
        this.left = left;
        this.right = right;
    }
 
    private void eat() {
        log.debug("eating...");
        Sleeper.sleep(1);
    }
 
    @Override
    public void run() {
        while (true) {
            // 获得左手筷子
            synchronized (left) {
                // 获得右手筷子
                synchronized (right) {
                    // 吃饭
                    eat();
                }
                // 放下右手筷子
            }
            // 放下左手筷子
        }
    }
}

就餐：

Chopstick c1 = new Chopstick("1");
Chopstick c2 = new Chopstick("2");
Chopstick c3 = new Chopstick("3");
Chopstick c4 = new Chopstick("4");
Chopstick c5 = new Chopstick("5");
new Philosopher("苏格拉底", c1, c2).start();
new Philosopher("柏拉图", c2, c3).start();
new Philosopher("亚里士多德", c3, c4).start();
new Philosopher("赫拉克利特", c4, c5).start();
new Philosopher("阿基米德", c5, c1).start();

五位哲学家与五根筷子围成一桌，每人需要左右手两根筷子才能吃饭

执行不多会儿就执行不下去了：

12:33:15.575 [苏格拉底] c.Philosopher - eating...
12:33:15.575 [亚里士多德] c.Philosopher - eating...
12:33:16.580 [阿基米德] c.Philosopher - eating...
12:33:17.580 [阿基米德] c.Philosopher - eating...
// 卡在这里, 不向下运行

用 jconsole 检测死锁，发现五个人各自拿着一根筷子、等待右边的筷子，形成了循环等待：

-------------------------------------------------------------------------
名称: 阿基米德
状态: Chopstick@1540e19d (筷子1) 上的BLOCKED, 拥有者: 苏格拉底
总阻止数: 2, 总等待数: 1
 
堆栈跟踪:
Philosopher.run(TestDinner.java:48)
   - 已锁定 Chopstick@6d6f6e28 (筷子5)
-------------------------------------------------------------------------
名称: 苏格拉底
状态: Chopstick@677327b6 (筷子2) 上的BLOCKED, 拥有者: 柏拉图
堆栈跟踪:
Philosopher.run(TestDinner.java:48)
   - 已锁定 Chopstick@1540e19d (筷子1)
-------------------------------------------------------------------------
（其余三位哲学家同样互相等待）

活锁

活锁出现在两个线程互相改变对方的结束条件，最后谁也无法结束。例如：

public class TestLiveLock {
    static volatile int count = 10;
    static final Object lock = new Object();
 
    public static void main(String[] args) {
        new Thread(() -> {
            // 期望减到 0 退出循环
            while (count > 0) {
                sleep(0.2);
                count--;
                log.debug("count: {}", count);
            }
        }, "t1").start();
        new Thread(() -> {
            // 期望超过 20 退出循环
            while (count < 20) {
                sleep(0.2);
                count++;
                log.debug("count: {}", count);
            }
        }, "t2").start();
    }
}

死锁是线程都阻塞住「不动」了，活锁则是线程一直在运行，却互相使对方的条件永远无法满足。可以加入随机退避时间来避免活锁。

饥饿

很多教程把饥饿定义为：一个线程由于优先级太低，始终得不到 CPU 调度执行，也不能够结束。饥饿的情况不易演示，讲读写锁时会涉及饥饿问题。

使用顺序加锁的方式可以解决哲学家就餐类的死锁问题：让所有线程都按照同一个全局顺序申请锁（比如所有哲学家都先拿编号小的筷子），循环等待就不会形成。但顺序加锁也可能带来饥饿：某些线程可能长时间抢不到排在后面的锁。

4.13 ReentrantLock

相对于 synchronized，ReentrantLock 具备如下特点：

可中断
可以设置超时时间
可以设置为公平锁
支持多个条件变量

与 synchronized 一样，它也支持可重入。

基本语法

// 获取锁
reentrantLock.lock();
try {
    // 临界区
} finally {
    // 释放锁
    reentrantLock.unlock();
}

与 synchronized 不同，释放锁的操作必须由开发者在 finally 中显式完成。

可重入

可重入是指同一个线程如果首次获得了这把锁，那么因为它是这把锁的拥有者，因此有权利再次获取这把锁。如果是不可重入锁，第二次获得锁时自己也会被自己挡住。

static ReentrantLock lock = new ReentrantLock();
 
public static void main(String[] args) {
    method1();
}
 
public static void method1() {
    lock.lock();
    try {
        log.debug("execute method1");
        method2();
    } finally {
        lock.unlock();
    }
}
 
public static void method2() {
    lock.lock();
    try {
        log.debug("execute method2");
        method3();
    } finally {
        lock.unlock();
    }
}
 
public static void method3() {
    lock.lock();
    try {
        log.debug("execute method3");
    } finally {
        lock.unlock();
    }
}

输出：

17:59:11.862 [main] c.TestReentrant - execute method1
17:59:11.865 [main] c.TestReentrant - execute method2
17:59:11.865 [main] c.TestReentrant - execute method3

可打断

使用 lockInterruptibly() 加锁，在等锁的过程中可以被 interrupt 打断：

ReentrantLock lock = new ReentrantLock();
 
Thread t1 = new Thread(() -> {
    log.debug("启动...");
    try {
        lock.lockInterruptibly();
    } catch (InterruptedException e) {
        e.printStackTrace();
        log.debug("等锁的过程中被打断");
        return;
    }
    try {
        log.debug("获得了锁");
    } finally {
        lock.unlock();
    }
}, "t1");
 
 
lock.lock();
log.debug("获得了锁");
t1.start();
try {
    sleep(1);
    t1.interrupt();
    log.debug("执行打断");
} finally {
    lock.unlock();
}

输出：

18:02:40.520 [main] c.TestInterrupt - 获得了锁
18:02:40.524 [t1] c.TestInterrupt - 启动...
18:02:41.530 [main] c.TestInterrupt - 执行打断
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(...)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(...)
    at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
    at TestInterrupt.lambda$main$0(TestInterrupt.java:17)
    at java.lang.Thread.run(Thread.java:748)
18:02:41.532 [t1] c.TestInterrupt - 等锁的过程中被打断

注意：如果使用的是不可中断模式 lock()，即使调用了 interrupt 也不会让等待中断：

ReentrantLock lock = new ReentrantLock();
 
Thread t1 = new Thread(() -> {
    log.debug("启动...");
    lock.lock();
    try {
        log.debug("获得了锁");
    } finally {
        lock.unlock();
    }
}, "t1");
 
 
lock.lock();
log.debug("获得了锁");
t1.start();
try {
    sleep(1);
    t1.interrupt();
    log.debug("执行打断");
    sleep(1);
} finally {
    log.debug("释放了锁");
    lock.unlock();
}

输出：

18:06:56.261 [main] c.TestInterrupt - 获得了锁
18:06:56.265 [t1] c.TestInterrupt - 启动...
18:06:57.266 [main] c.TestInterrupt - 执行打断 // 这时 t1 并没有被真正打断, 而是仍继续等待锁
18:06:58.267 [main] c.TestInterrupt - 释放了锁
18:06:58.267 [t1] c.TestInterrupt - 获得了锁

锁超时

立刻失败：使用无参的 tryLock()，获取不到锁立刻返回 false。

ReentrantLock lock = new ReentrantLock();
Thread t1 = new Thread(() -> {
    log.debug("启动...");
    if (!lock.tryLock()) {
        log.debug("获取立刻失败，返回");
        return;
    }
    try {
        log.debug("获得了锁");
    } finally {
        lock.unlock();
    }
}, "t1");
 
lock.lock();
log.debug("获得了锁");
t1.start();
try {
    sleep(2);
} finally {
    lock.unlock();
}

输出：

18:15:02.918 [main] c.TestTimeout - 获得了锁
18:15:02.921 [t1] c.TestTimeout - 启动...
18:15:02.921 [t1] c.TestTimeout - 获取立刻失败，返回

超时失败：使用 tryLock(long, TimeUnit)，等待一段时间仍获取不到就失败返回。

ReentrantLock lock = new ReentrantLock();
Thread t1 = new Thread(() -> {
    log.debug("启动...");
    try {
        if (!lock.tryLock(1, TimeUnit.SECONDS)) {
            log.debug("获取等待 1s 后失败，返回");
            return;
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    try {
        log.debug("获得了锁");
    } finally {
        lock.unlock();
    }
}, "t1");
 
lock.lock();
log.debug("获得了锁");
t1.start();
try {
    sleep(2);
} finally {
    lock.unlock();
}

输出：

18:19:40.537 [main] c.TestTimeout - 获得了锁
18:19:40.544 [t1] c.TestTimeout - 启动...
18:19:41.547 [t1] c.TestTimeout - 获取等待 1s 后失败，返回

使用 tryLock 解决哲学家就餐问题

让筷子继承 ReentrantLock，哲学家拿不到第二根筷子时放下已拿到的筷子并稍后重试，破坏死锁条件：

class Chopstick extends ReentrantLock {
    String name;
 
    public Chopstick(String name) {
        this.name = name;
    }
 
    @Override
    public String toString() {
        return "筷子{" + name + '}';
    }
}
 
class Philosopher extends Thread {
    Chopstick left;
    Chopstick right;
 
    public Philosopher(String name, Chopstick left, Chopstick right) {
        super(name);
        this.left = left;
        this.right = right;
    }
 
    @Override
    public void run() {
        while (true) {
            // 尝试获得左手筷子
            if (left.tryLock()) {
                try {
                    // 尝试获得右手筷子
                    if (right.tryLock()) {
                        try {
                            eat();
                        } finally {
                            right.unlock();
                        }
                    }
                } finally {
                    left.unlock();
                }
            }
        }
    }
 
    private void eat() {
        log.debug("eating...");
        Sleeper.sleep(1);
    }
}

公平锁

ReentrantLock 默认是不公平的：

ReentrantLock lock = new ReentrantLock(false);
 
lock.lock();
for (int i = 0; i < 500; i++) {
    new Thread(() -> {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + " running...");
        } finally {
            lock.unlock();
        }
    }, "t" + i).start();
}
 
// 1s 之后去争抢锁
Thread.sleep(1000);
new Thread(() -> {
    System.out.println(Thread.currentThread().getName() + " start...");
    lock.lock();
    try {
        System.out.println(Thread.currentThread().getName() + " running...");
    } finally {
        lock.unlock();
    }
}, "强行插入").start();
lock.unlock();

非公平锁下，后到的线程有机会「强行插入」，在队列中间输出（该实验不一定总能复现）：

t39 running...
t40 running...
t41 running...
t42 running...
t43 running...
强行插入 start...
强行插入 running...
t44 running...
t45 running...

改为公平锁后：

ReentrantLock lock = new ReentrantLock(true);

「强行插入」线程总是排在最后输出：

t465 running...
t464 running...
t477 running...
t442 running...
t468 running...
t493 running...
t482 running...
t485 running...
t481 running...
强行插入 running...

公平锁一般没有必要，会降低并发度，后面分析 AQS 原理时会讲解。

条件变量

synchronized 中也有条件变量，就是原理中的 WaitSet 休息室，当条件不满足时进入 WaitSet 等待。ReentrantLock 的条件变量比 synchronized 强大之处在于，它支持多个条件变量：

synchronized 是所有不满足条件的线程都在一间休息室等消息；
ReentrantLock 支持多间休息室，有专门等烟的休息室、专门等早餐的休息室，唤醒时也按休息室来唤醒。

使用要点：

await 前需要获得锁；
await 执行后会释放锁，进入 conditionObject 等待；
await 的线程被唤醒（或打断、或超时）后重新竞争 lock 锁；
竞争 lock 锁成功后，从 await 后继续执行。

例子：

static ReentrantLock lock = new ReentrantLock();
static Condition waitCigaretteQueue = lock.newCondition();
static Condition waitbreakfastQueue = lock.newCondition();
static volatile boolean hasCigrette = false;
static volatile boolean hasBreakfast = false;
 
public static void main(String[] args) {
    new Thread(() -> {
        try {
            lock.lock();
            while (!hasCigrette) {
                try {
                    waitCigaretteQueue.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("等到了它的烟");
        } finally {
            lock.unlock();
        }
    }).start();
 
    new Thread(() -> {
        try {
            lock.lock();
            while (!hasBreakfast) {
                try {
                    waitbreakfastQueue.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("等到了它的早餐");
        } finally {
            lock.unlock();
        }
    }).start();
 
    sleep(1);
    sendBreakfast();
    sleep(1);
    sendCigarette();
}
 
private static void sendCigarette() {
    lock.lock();
    try {
        log.debug("送烟来了");
        hasCigrette = true;
        waitCigaretteQueue.signal();
    } finally {
        lock.unlock();
    }
}
 
private static void sendBreakfast() {
    lock.lock();
    try {
        log.debug("送早餐来了");
        hasBreakfast = true;
        waitbreakfastQueue.signal();
    } finally {
        lock.unlock();
    }
}

输出：

18:52:27.680 [main] c.TestCondition - 送早餐来了
18:52:27.682 [Thread-1] c.TestCondition - 等到了它的早餐
18:52:28.683 [main] c.TestCondition - 送烟来了
18:52:28.683 [Thread-0] c.TestCondition - 等到了它的烟

不同条件的等待线程分别在不同的 Condition 上等待，signal 时只会唤醒对应休息室中的线程，比 notifyAll 一锅端式的唤醒更加精确。