全面深入学习Java并发编程-第五章

文章目录

一、wait & notify
- 1. 小故事理解 wait & notify
- 2. wait & notify 原理
- 3. wait & notify API
- 4. sleep 与 wait 区别
- 5. wait & notify 使用套路
二、多线程设计模式之保护性暂停（同步模式）
- 1. 定义
- 2. 实现
- 3. join 原理
三、多线程设计模式之生产者/消费者（异步模式）
- 1. 定义
- 2. 实现
四、Park & Unpark
- 1. 基本使用
- 2. 特点
- 3. 原理
五、线程活跃性
- 1. 死锁
- - （1）定位死锁
  - （2）哲学家就餐问题
- 2. 活锁
- 3. 饥饿
六、ReentrantLock
- 1. ReentrantLock 与 synchronized 比较
- 2. ReentrantLock 特点
- - （1）可重入
  - （2）可中断
  - （3）锁超时
  - - - 锁超时-解决哲学家就餐问题
  - （4）可设为公平锁
  - （5）支持多个条件变量
七、多线程设计模式之控制线程顺序（同步模式）
- 1. 固定顺序
- - （1） wait & notify
  - （2） park & unpark
  - （3）ReentrantLock使用条件变量Condition
- 2. 交替顺序
- - （1）wait & notify
  - （2）park & unpark
  - （3）ReentrantLock使用条件变量Condition

一、wait & notify 1. 小故事理解 wait & notify

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

小南于是可以离开休息室，重新进入竞争锁的队列

2. wait & notify 原理

Owner 线程发现条件不满足，调用 wait 方法，即可进入 WaitSet 变为 WAITING 状态
BLOCKED 和 WAITING 的线程都处于阻塞状态，不占用 CPU 时间片
BLOCKED 线程会在 Owner 线程释放锁时被唤醒
WAITING 线程会在 Owner 线程调用 notify 或 notifyAll 时被唤醒，但唤醒后并不意味者立刻获得锁，仍需进入EntryList 重新竞争

3. wait & notify API

obj.wait() 让进入 object 监视器的线程到 waitSet 等待
wait() 方法会释放对象的锁，进入 WaitSet 等待区，从而让其他线程就机会获取对象的锁。无限制等待，直到被 notify 为止
wait(long n) 有时限的等待, 到 n 毫秒后结束等待，或是被 notify 提前唤醒
obj.notify() 在 object 上正在 waitSet 等待的线程中挑一个唤醒
obj.notifyAll() 让 object 上正在 waitSet 等待的线程全部唤醒

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

@Slf4j(topic = "c.TestWaitNotify")
public class TestWaitNotify {
    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("t1被唤醒，执行其它代码....");
            }
        },"t1").start();

        new Thread(() -> {
            synchronized (obj) {
                log.debug("执行....");
                try {
                    obj.wait(); // 让线程在obj上一直等待下去
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("t2被唤醒，执行其它代码....");
            }
        },"t2").start();

        // 主线程两秒后执行
        sleep(0.5);
        log.debug("唤醒 obj 上所有线程");
        synchronized (obj) {
//            obj.notify(); // 唤醒obj上一个线程
            obj.notifyAll(); // 唤醒obj上所有等待线程
        }
    }
}

14:44:13.682 c.TestWaitNotify [t1] - 执行....
14:44:13.684 c.TestWaitNotify [t2] - 执行....
14:44:14.194 c.TestWaitNotify [main] - 唤醒 obj 上所有线程
14:44:14.194 c.TestWaitNotify [t2] - t2被唤醒，执行其它代码....
14:44:14.194 c.TestWaitNotify [t1] - t1被唤醒，执行其它代码....

4. sleep 与 wait 区别

（1）sleep 是 Thread 方法，而 wait 是 Object 的方法
（2） sleep 不需要强制和 synchronized 配合使用，但 wait 需要和 synchronized 一起用
（3） sleep 在睡眠的同时，不会释放对象锁的，但 wait 在等待的时候会释放对象锁
（4）它们状态都是 TIMED_WAITING

5. wait & notify 使用套路

// wait 线程
new Thread(() -> {
	synchronized(lock) {
	// 用 while + wait，当条件不成立时，可以再次 wait
	 while(条件不成立) {
	 	lock.wait();
	 }
	 // 被唤醒后再干活
	}
}, "t1").start();

// notifyAll 线程
new Thread(() -> {
	synchronized(lock) {
	 	lock.notifyAll();
	}
}, "t2").start();

案例

@Slf4j(topic = "c.TestCorrectPosture")
public class TestCorrectPostureStep5 {
    static final Object room = new Object(); // 对象锁最好使用final修饰，防止改变
    static boolean hasCigarette = false; // 是否有烟
    static boolean hasTakeout = false; // 是否有外卖

    public static void main(String[] args) throws InterruptedException {


        new Thread(() -> {
            synchronized (room) {
                log.debug("有烟没？[{}]", hasCigarette);
                while (!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);
                while (!hasTakeout) {
                    log.debug("没外卖，先歇会！");
                    try {
                        room.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                log.debug("外卖送到没？[{}]", hasTakeout);
                if (hasTakeout) {
                    log.debug("可以开始干活了");
                } else {
                    log.debug("没干成活...");
                }
            }
        }, "小女").start();

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


    }

}

20:14:17.024 c.TestCorrectPosture [小南] - 有烟没？[false]
20:14:17.042 c.TestCorrectPosture [小南] - 没烟，先歇会！
20:14:17.042 c.TestCorrectPosture [小女] - 外卖送到没？[false]
20:14:17.043 c.TestCorrectPosture [小女] - 没外卖，先歇会！
20:14:18.029 c.TestCorrectPosture [送外卖的] - 外卖到了噢！
20:14:18.029 c.TestCorrectPosture [小女] - 外卖送到没？[true]
20:14:18.029 c.TestCorrectPosture [小女] - 可以开始干活了
20:14:18.029 c.TestCorrectPosture [小南] - 没烟，先歇会！

二、多线程设计模式之保护性暂停（同步模式） 1. 定义

即 Guarded Suspension，用在一个线程同步等待另一个线程的执行结果

要点

有一个结果需要从一个线程传递到另一个线程，让他们关联同一个 GuardedObject
如果有结果不断从一个线程到另一个线程那么可以使用消息队列（见生产者/消费者）
JDK 中，join 的实现、Future 的实现，采用的就是此模式
因为要等待另一方的结果，因此归类到同步模式

2. 实现

@Slf4j(topic = "c.TestGuardedObject")
public class TestGuardedObject {
    public static void main(String[] args) {

        GuardedObject guardedObject = new GuardedObject();
        // t1等待获取结果，超过2s则退出
        new Thread(() -> {
            log.debug("begin");
            Object response = guardedObject.get(2000);
            log.debug("获取的结果：{}", response);
        }, "t1").start();

        // t2 完成后，通知 t1
        new Thread(() -> {
            log.debug("begin");
            try {
            	// t2 延迟 1s 返回结果 
                TimeUnit.SECONDS.sleep(1);
                // TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            guardedObject.complete(new Object());
        }, "t2").start();
    }


}

@Slf4j(topic = "c.GuardedObject")
class GuardedObject {
    // 存放结果
    private Object response;

    private final Object lock = new Object();

    // 获取结果
    // timeout 表示等待的超时时间
    public Object get(long timeout) {
        synchronized (lock) {
            // 1) 记录最初时间
            long begin = System.currentTimeMillis();
            // 2) 已经经历的时间
            long timePassed = 0;
            while (response == null) {
                // 4) 假设 timeout 是 10s，结果在 4s 时被唤醒了，那么还有 6s 要等
                long waitTime = timeout - timePassed;
                if (waitTime <= 0) {
                    log.debug("超过等待的最大时间，退出...");
                    break;
                }
                try {
                    lock.wait(waitTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 3) 如果提前被唤醒，这时已经经历的时间假设为 4s
                timePassed = System.currentTimeMillis() - begin;
            }
            return response;
        }
    }

    public void complete(Object response) {
        synchronized (lock) {
            // 条件满足，通知等待线程
            this.response = response;
            log.debug("notify...");
            lock.notifyAll();
        }
    }

}

t1 等待结果超时时间是2s，当 t2 延迟 1s 返回结果，TimeUnit.SECONDS.sleep(1);

22:15:11.447 c.TestGuardedObject [t2] - begin
22:15:11.449 c.TestGuardedObject [t1] - begin
22:15:12.452 c.GuardedObject [t2] - notify...
22:15:12.452 c.TestGuardedObject [t1] - 获取的结果：java.lang.Object@647f76d5

t1 等待结果超时时间是2s，当 t2 延迟 3s 返回结果，TimeUnit.SECONDS.sleep(3);

22:17:38.536 c.TestGuardedObject [t2] - begin
22:17:38.541 c.TestGuardedObject [t1] - begin
22:17:40.541 c.GuardedObject [t1] - 超过等待的最大时间，退出...
22:17:40.541 c.TestGuardedObject [t1] - 获取的结果：null
22:17:41.541 c.GuardedObject [t2] - notify...

3. join 原理

join 体现的是【保护性暂停】模式
t1.join(); 调用者轮询检查t1线程 alive 状态
实际调用的是 wait 方法

join 源码

public final synchronized void join(long millis)
    throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;

        if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
        }

        if (millis == 0) {
        	// 调用者轮询检查线程 alive 状态
            while (isAlive()) {
                wait(0);
            }
        } else {
        	// 调用者轮询检查线程 alive 状态
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                wait(delay);
                // 防止虚假唤醒，记录已经等待的时间
                now = System.currentTimeMillis() - base;
            }
        }
    }

t1.join();
等价于下面的代码
synchronized (t1) {
	// 调用者线程进入 t1 的 waitSet 等待, 直到 t1 运行结束
	while (t1.isAlive()) {
		t1.wait(0);
	}
}

三、多线程设计模式之生产者/消费者（异步模式）

1. 定义

要点

与前面的保护性暂停中的 GuardObject 不同，不需要产生结果和消费结果的线程一一对应
消费队列可以用来平衡生产和消费的线程资源
生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
JDK 中各种阻塞队列，采用的就是这种模式

2. 实现

@Slf4j(topic = "c.Test21")
public class Test21 {

    public static void main(String[] args) {
        MessageQueue queue = new MessageQueue(2);

        for (int i = 0; i < 3; i++) {
            int id = i;
            new Thread(() -> {
                queue.put(new Message(id , "值"+id));
            }, "生产者" + i).start();
        }

        new Thread(() -> {
            while(true) {
                sleep(1);
                Message message = queue.take();
            }
        }, "消费者").start();
    }

}

// 消息队列类 ， java 线程之间通信
@Slf4j(topic = "c.MessageQueue")
class MessageQueue {
    // 消息的队列集合，双向队列
    private LinkedList list = new LinkedList<>();
    // 队列容量
    private int capcity;

    public MessageQueue(int capcity) {
        this.capcity = capcity;
    }

    // 获取消息
    public Message take() {
        // 检查队列是否为空
        synchronized (list) {
            while(list.isEmpty()) {
                try {
                    log.debug("队列为空, 消费者线程等待");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 从队列头部获取消息并返回
            Message message = list.removeFirst();
            log.debug("已消费消息 {}", message);
            list.notifyAll();
            return message;
        }
    }

    // 存入消息
    public void put(Message message) {
        synchronized (list) {
            // 检查对象是否已满
            while(list.size() == capcity) {
                try {
                    log.debug("队列已满, 生产者线程等待");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 将消息加入队列尾部
            list.addLast(message);
            log.debug("已生产消息 {}", message);
            list.notifyAll();
        }
    }
}

@Data
@AllArgsConstructor
final class Message {
    private int id;
    private Object value;
}

10:59:44.966 c.MessageQueue [生产者0] - 已生产消息 Message(id=0, value=值0)
10:59:44.975 c.MessageQueue [生产者2] - 已生产消息 Message(id=2, value=值2)
10:59:44.975 c.MessageQueue [生产者1] - 队列已满, 生产者线程等待
10:59:45.976 c.MessageQueue [消费者] - 已消费消息 Message(id=0, value=值0)
10:59:45.976 c.MessageQueue [生产者1] - 已生产消息 Message(id=1, value=值1)
10:59:46.990 c.MessageQueue [消费者] - 已消费消息 Message(id=2, value=值2)
10:59:48.002 c.MessageQueue [消费者] - 已消费消息 Message(id=1, value=值1)
10:59:49.009 c.MessageQueue [消费者] - 队列为空, 消费者线程等待

四、Park & Unpark 1. 基本使用

它们是 LockSupport 类中的方法

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

案例一：先 park 再 unpark

@Slf4j(topic = "c.TestParkUnpark")
public class TestParkUnpark {
    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            log.debug("start...");
            // 先 park 再 unpark， t1线程休眠时间小于main线程休眠时间
            sleep(2);
            log.debug("t1线程先park...");
            LockSupport.park();
            log.debug("resume...");
        }, "t1");
        t1.start();


        sleep(3);
        log.debug("unpark...");
        LockSupport.unpark(t1);
    }
}

11:12:17.676 c.TestParkUnpark [t1] - start...
11:12:19.681 c.TestParkUnpark [t1] - t1线程先park...
11:12:20.676 c.TestParkUnpark [main] - unpark...
11:12:20.676 c.TestParkUnpark [t1] - resume...

案例二：先 unpark 再 park

先 unpark 后，就算再 park，也不会阻塞住

@Slf4j(topic = "c.TestParkUnpark")
public class TestParkUnpark {
    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            log.debug("start...");
            sleep(2);
            log.debug("park...");
            LockSupport.park();
            log.debug("resume...");
        }, "t1");
        t1.start();

        // 先 unpark 再 park ， main线程休眠时间小于t1线程休眠时间
        sleep(1);
        log.debug("main线程先unpark...");
        LockSupport.unpark(t1);
    }
}

11:09:13.101 c.TestParkUnpark [t1] - start...
11:09:14.111 c.TestParkUnpark [main] - main线程先unpark...
11:09:15.113 c.TestParkUnpark [t1] - park...
11:09:15.113 c.TestParkUnpark [t1] - resume...

2. 特点

与 Object 的 wait & notify 相比

wait，notify 和 notifyAll 必须配合 Object Monitor 一起使用，而 park，unpark 不需要
park & unpark 是以线程为单位来【阻塞】和【唤醒】线程（精确唤醒对应线程），而 notify 只能随机唤醒一个等待线程，notifyAll是唤醒所有等待线程，就不那么【精确】
park & unpark 可以先 unpark，而 wait & notify 不能先 notify

3. 原理

每个线程都有自己的一个 Parker 对象，由三部分组成 _counter （计数器）， _cond（条件变量）和 _mutex（互斥锁）。Parker对象是底层C语言实现的。
调用park时，首先会检查需不需要阻塞：
_counter = 1 运行，_counter = 0 阻塞。
调用unpark时，会让_counter = 1：
- 如果线程此时正在阻塞，则唤醒线程继续运行；
- 如果线程此时没有阻塞，则下次调用park时，发现_counter = 1，线程不会阻塞而是继续运行；
多次调用 unpark，_counter 仍为1，不会自增

（1）分析先 park 再 unpark

当前线程调用 Unsafe.park() 方法
检查 _counter ，本情况为 0，这时，获得 _mutex 互斥锁
线程进入 _cond 条件变量阻塞
设置 _counter = 0
调用 Unsafe.unpark(Thread_0) 方法，设置 _counter 为 1
唤醒 _cond 条件变量中的 Thread_0
Thread_0 恢复运行
设置 _counter 为 0

（2）分析先 unpark 再 park

调用 Unsafe.unpark(Thread_0) 方法，设置 _counter 为 1
当前线程调用 Unsafe.park() 方法
检查 _counter ，本情况为 1，这时线程无需阻塞，继续运行
设置 _counter 为 0

五、线程活跃性

线程活跃性是线程的代码有限但一直执行不完。
线程活跃性包含死锁、活锁、饥饿

1. 死锁

一个线程需要同时获取多把锁，这时就容易发生死锁
线程之间互相持有对方的锁，导致都无法继续运行，线程被阻塞住；（顺序加锁可以解决死锁问题，但可能会产生饥饿问题）

例如：

t1 线程获得 A对象锁，接下来想获取 B对象的锁
t2 线程获得 B对象锁，接下来想获取 A对象的锁

@Slf4j(topic = "c.TestDeadLock")
public class TestDeadLock {
    public static void main(String[] args) {
        test1();
    }

    private static void test1() {
        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();
    }
}

发生死锁，程序停止不了

死锁分析

顺序加锁的解决方案

线程A 按 A、B 加锁
线程B 也按 A、B 加锁

（1）定位死锁

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

详情见第二章（查看进程和线程的方法/java）通过 jstack 或 jconsole 查看线程状态

（2）哲学家就餐问题

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

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

public class TestDeadLock {
    public static void main(String[] args) {
        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();
        // 按c5, c1顺序会出现死锁，按c1, c5虽然不会出现死锁，但会出现饥饿
        new Philosopher("阿基米德", c5, c1).start();
    }
}

@Slf4j(topic = "c.Philosopher")
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) {
            //　尝试获得左手筷子
            synchronized (left) {
                // 尝试获得右手筷子
                synchronized (right) {
                    eat();
                }
            }
        }
    }

    Random random = new Random();
    private void eat() {
        log.debug("eating...");
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class Chopstick {
    String name;

    public Chopstick(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "筷子{" + name + '}';
    }
}

17:37:15.858 c.Philosopher [亚里士多德] - eating...
17:37:15.858 c.Philosopher [苏格拉底] - eating...
17:37:16.368 c.Philosopher [苏格拉底] - eating...
17:37:16.368 c.Philosopher [亚里士多德] - eating...
17:37:16.880 c.Philosopher [阿基米德] - eating...
// 卡在这里, 不向下运行

使用 jconsole 检测死锁
这种线程没有按预期结束，执行不下去的情况，归类为【活跃性】问题

-------------------------------------------------------------------------
名称: 阿基米德
状态: cn.itcast.Chopstick@1540e19d (筷子1) 上的BLOCKED, 拥有者: 苏格拉底
总阻止数: 2, 总等待数: 1
堆栈跟踪:
cn.itcast.Philosopher.run(TestDinner.java:48)
- 已锁定 cn.itcast.Chopstick@6d6f6e28 (筷子5)
-------------------------------------------------------------------------
名称: 苏格拉底
状态: cn.itcast.Chopstick@677327b6 (筷子2) 上的BLOCKED, 拥有者: 柏拉图
总阻止数: 2, 总等待数: 1
堆栈跟踪:
cn.itcast.Philosopher.run(TestDinner.java:48)
- 已锁定 cn.itcast.Chopstick@1540e19d (筷子1)
-------------------------------------------------------------------------
名称: 柏拉图
状态: cn.itcast.Chopstick@14ae5a5 (筷子3) 上的BLOCKED, 拥有者: 亚里士多德
总阻止数: 2, 总等待数: 0
堆栈跟踪:
cn.itcast.Philosopher.run(TestDinner.java:48)
- 已锁定 cn.itcast.Chopstick@677327b6 (筷子2)
-------------------------------------------------------------------------
名称: 亚里士多德
状态: cn.itcast.Chopstick@7f31245a (筷子4) 上的BLOCKED, 拥有者: 赫拉克利特
总阻止数: 1, 总等待数: 1
堆栈跟踪:
cn.itcast.Philosopher.run(TestDinner.java:48)
- 已锁定 cn.itcast.Chopstick@14ae5a5 (筷子3)
-------------------------------------------------------------------------
名称: 赫拉克利特
状态: cn.itcast.Chopstick@6d6f6e28 (筷子5) 上的BLOCKED, 拥有者: 阿基米德
总阻止数: 2, 总等待数: 0
堆栈跟踪:
cn.itcast.Philosopher.run(TestDinner.java:48)
- 已锁定 cn.itcast.Chopstick@7f31245a (筷子4)

2. 活锁

活锁：线程没有被阻塞住，但是线程之间改变了对方的结束条件，导致都无法结束；（解决活锁可以设置随机休眠时间使线程交错运行）

@Slf4j(topic = "c.TestLiveLock")
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();
    }
}

停不下来

3. 饥饿

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

六、ReentrantLock 1. ReentrantLock 与 synchronized 比较

ReentrantLock 实现了 Lock 接口，Lock锁底层实现原理：AQS+CAS+LockSupport实现
Lock锁基于AQS+CAS+LockSupport实现，但是Lock锁的升级过程需要自己实现
synchronized基于C++实现，JDK1.6之后可以实现锁的升级
ConcurrentHashMap1.7用的是Lock锁，1.8用的是synchronized
ReentrantLock是对象锁，synchronized是关键字
synchronized、ReentrantLock都是可重入的

ReentrantLock 基本语法

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

2. ReentrantLock 特点（1）可重入

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

@Slf4j(topic = "c.TestReentrant")
public class TestReentrant {
    static ReentrantLock lock = new ReentrantLock();

    public static void main(String[] args) {
        method1();
    }

    public static void method1() {
        lock.lock();
        try {
            log.debug("enter method1");
            method2();
        } finally {
            lock.unlock();
        }
    }

    public static void method2() {
        lock.lock();
        try {
            log.debug("enter method2");
            method3();
        } finally {
            lock.unlock();
        }
    }

    public static void method3() {
        lock.lock();
        try {
            log.debug("enter method3");
        } finally {
            lock.unlock();
        }
    }
}

18:06:01.477 c.TestReentrant [main] - enter method1
18:06:01.479 c.TestReentrant [main] - enter method2
18:06:01.480 c.TestReentrant [main] - enter method3

（2）可中断

调用lock.lockInterruptibly()，是可打断的，lock.lock()是不可打断的
如果是不可中断模式，那么即使使用了 interrupt 也不会让等待中断
可打断是被动的避免死锁，需要其他线程调用interrupt来打断线程

@Slf4j(topic = "c.TestInterrupt")
public class TestInterrupt {
    public static void main(String[] args) {
        test1();
    }

    private static void test1() {
        ReentrantLock lock = new ReentrantLock();

        Thread t1 = new Thread(() -> {
            log.debug("启动t1线程...");
            try {
                lock.lockInterruptibly();
            } catch (InterruptedException e) {
                e.printStackTrace();
                // t1被打断时还没有获取到锁，会进入中断异常，直接return，不需要释放锁
                log.debug("t1等锁的过程中被打断");
                return;
            }
            try {
                log.debug("t1获得了锁");
            } finally {
                lock.unlock();
            }
        }, "t1");

        // 先让main线程获得锁
        lock.lock();
        log.debug("main获得了锁");
        // 再启动t1线程
        t1.start();
        try {
            sleep(1);
            t1.interrupt();
            log.debug("main线程打断t1线程");
        } finally {
            lock.unlock();
        }
    }
}

18:20:50.856 c.TestInterrupt [main] - main获得了锁
18:20:50.859 c.TestInterrupt [t1] - 启动t1线程...
18:20:51.875 c.TestInterrupt [main] - main线程打断t1线程
18:20:51.876 c.TestInterrupt [t1] - t1等锁的过程中被打断
java.lang.InterruptedException
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchronizer.java:898)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1222)
	at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
	at cn.itcast.n4.reentrant.TestInterrupt.lambda$test1$1(TestInterrupt.java:49)
	at java.lang.Thread.run(Thread.java:748)

（3）锁超时

锁超时是主动避免死锁，超过一定时间自动获取锁失败退出
lock.tryLock(超时时间, TimeUnit.SECONDS) 表示超时失败
lock.tryLock() 表示立刻失败
lock.tryLock() 支持被打断

@Slf4j(topic = "c.TestTimeout")
public class TestTimeout {
    public static void main(String[] args) {
        test1();
    }

    private static void test1() {
        ReentrantLock lock = new ReentrantLock();
        Thread t1 = new Thread(() -> {
            log.debug("启动t1线程...");
            try {
                if (!lock.tryLock(1, TimeUnit.SECONDS)) {
                    log.debug("t1等待 1s 后获取锁失败，返回");
                    return;
                }
            } catch (InterruptedException e) {
                // lock.tryLock 也支持被打断
                e.printStackTrace();
                return;
            }
            try {
                log.debug("t1获得了锁");
            } finally {
                lock.unlock();
            }
        }, "t1");

        lock.lock();
        log.debug("main获得锁");
        t1.start();
        try {
            sleep(2);
            log.debug("main休眠完，释放锁");
        } finally {
            lock.unlock();
        }
    }
 }

18:34:03.067 c.TestTimeout [main] - main获得锁
18:34:03.072 c.TestTimeout [t1] - 启动t1线程...
18:34:04.074 c.TestTimeout [t1] - t1等待 1s 后获取锁失败，返回
18:34:05.080 c.TestTimeout [main] - main休眠完，释放锁

锁超时-解决哲学家就餐问题

public class TestDeadLock {
    public static void main(String[] args) {
        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();
    }
}

@Slf4j(topic = "c.Philosopher")
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 锁对象
class Chopstick extends ReentrantLock {
    String name;

    public Chopstick(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "筷子{" + name + '}';
    }
}

都等获取到就餐的机会

20:46:46.781 c.Philosopher [苏格拉底] - eating...
20:46:46.787 c.Philosopher [亚里士多德] - eating...
20:46:47.790 c.Philosopher [亚里士多德] - eating...
20:46:47.793 c.Philosopher [阿基米德] - eating...
20:46:48.819 c.Philosopher [柏拉图] - eating...
20:46:48.822 c.Philosopher [赫拉克利特] - eating...

（4）可设为公平锁

ReentrantLock lock = new ReentrantLock(true); // true为公平
ReentrantLock 默认是false不公平的
公平锁一般没有必要，会降低并发度

（5）支持多个条件变量

synchronized 中也有条件变量，就是Monitor中的 waitSet，当条件不满足时进入 waitSet 等待
ReentrantLock 支持多个条件变量
- synchronized 中不满足条件的线程都在同一个waitSet 等待
- ReentrantLock 支持每个线程有独立的waitSet，唤醒时可唤醒指定的 conditionObject

使用要点：

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

使用案例

@Slf4j(topic = "c.Test24")
public class Test24 {
    static final Object room = new Object();
    static boolean hasCigarette = false;
    static boolean hasTakeout = false;
    static ReentrantLock ROOM = new ReentrantLock();
    // 等待烟的休息室
    static Condition waitCigaretteSet = ROOM.newCondition();
    // 等外卖的休息室
    static Condition waitTakeoutSet = ROOM.newCondition();

    public static void main(String[] args) {


        new Thread(() -> {
            ROOM.lock();
            try {
                log.debug("有烟没？[{}]", hasCigarette);
                while (!hasCigarette) {
                    log.debug("没烟，先歇会！");
                    try {
                        waitCigaretteSet.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                log.debug("可以开始干活了");
            } finally {
                ROOM.unlock();
            }
        }, "小南").start();

        new Thread(() -> {
            ROOM.lock();
            try {
                log.debug("外卖送到没？[{}]", hasTakeout);
                while (!hasTakeout) {
                    log.debug("没外卖，先歇会！");
                    try {
                        waitTakeoutSet.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                log.debug("可以开始干活了");
            } finally {
                ROOM.unlock();
            }
        }, "小女").start();

        sleep(1);
        
        new Thread(() -> {
            ROOM.lock();
            try {
                hasTakeout = true;
                waitTakeoutSet.signal();
            } finally {
                ROOM.unlock();
            }
        }, "送外卖的").start();

        sleep(1);

        new Thread(() -> {
            ROOM.lock();
            try {
                hasCigarette = true;
                waitCigaretteSet.signal();
            } finally {
                ROOM.unlock();
            }
        }, "送烟的").start();
    }

}

21:10:05.387 c.Test24 [小南] - 有烟没？[false]
21:10:05.394 c.Test24 [小南] - 没烟，先歇会！
21:10:05.394 c.Test24 [小女] - 外卖送到没？[false]
21:10:05.395 c.Test24 [小女] - 没外卖，先歇会！
21:10:06.384 c.Test24 [小女] - 可以开始干活了
21:10:07.384 c.Test24 [小南] - 可以开始干活了

七、多线程设计模式之控制线程顺序（同步模式） 1. 固定顺序

要求：固定顺序先打印2后打印1

（1） wait & notify

①需要保证先 wait 再 notify，否则 wait 线程永远得不到唤醒。因此使用了『运行标记』来判断该不该wait；
②如果有些干扰线程错误地 notify 了 wait 线程（虚假唤醒），条件不满足时还要重新等待，使用了 while 循环来解决此问题；
③唤醒『同步对象』上的 wait 线程需要使用 notifyAll，因为『同步对象』上的等待线程可能不止一个；

public class Order_WaitNotify {
    // 『同步对象』锁对象
    static Object lock = new Object();
    // t2 运行标记， 代表 t2 是否执行过
    static boolean t2runed = false;

    public static void main(String[] args) {

        Thread t1 = new Thread(() -> {
            synchronized (lock) {
                // 如果 t2 没有执行过
                while (!t2runed) {
                    try {
                        // t1 先等一会
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            System.out.println(1);
        }, "t1");

        Thread t2 = new Thread(() -> {
            System.out.println(2);
            synchronized (lock) {
                // 修改运行标记
                t2runed = true;
                // 通知 lock 上等待的线程（可能有多个，因此需要用 notifyAll）
                lock.notifyAll();
            }
        }, "t2");

        t1.start();
        t2.start();
    }
}

[t2] - 2
[t1] - 1

（2） park & unpark

park 和 unpark 方法比较灵活，谁先调用，谁后调用无所谓。并且是以线程为单位进行『暂停』和『恢复』，不需要『同步对象』和『运行标记』。

public class Order_ParkUnpark {

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            // 当没有『许可』时，当前线程暂停运行；有『许可』时，用掉这个『许可』，当前线程恢复运行
            LockSupport.park();
            System.out.println("1");
        }, "t1");

        Thread t2 = new Thread(() -> {
            System.out.println("2");
            // 给线程 t1 发放『许可』（多次连续调用 unpark 只会发放一个『许可』）
            LockSupport.unpark(t1);
        }, "t2");

        t1.start();
        t2.start();
    }
}

[t2] - 2
[t1] - 1

（3）ReentrantLock使用条件变量Condition

public class Order_ConditionReentrantLock {

    static ReentrantLock lock = new ReentrantLock();
    static Condition condition1 = lock.newCondition();
    static Condition condition2 = lock.newCondition();
    static volatile boolean t1Flag;
    static volatile boolean t2Flag;

    public static void main(String[] args) {

        Thread t1 = new Thread(() -> {
            try {
                lock.lock();
                while (!t1Flag) {
                    condition1.await();
                }
                System.out.println(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }, "t1");

        Thread t2 = new Thread(() -> {
            try {
                lock.lock();
                while (!t2Flag) {
                    condition2.await();
                }
                System.out.println(2);
                t1Flag = true;
                condition1.signal();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }, "t2");

        t1.start();
        t2.start();

        // mian先唤醒t2，t2内部再去唤醒t1
        try {
            lock.lock();
            t2Flag = true;
            condition2.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

[t2] - 2
[t1] - 1

2. 交替顺序

要求：线程 1 输出 a 5 次，线程 2 输出 b 5 次，线程 3 输出 c 5 次。现在要求输出 abcabcabcabcabc

（1）wait & notify

public class SwapOrder_WaitNotify {

    public static void main(String[] args) {
        // 从a线程开始打印，交替打印5轮
        SyncWaitNotify syncWaitNotify = new SyncWaitNotify("a", 5);
        new Thread(() -> {
            syncWaitNotify.print("a", "b");
        },"a").start();
        new Thread(() -> {
            syncWaitNotify.print("b", "c");
        },"b").start();
        new Thread(() -> {
            syncWaitNotify.print("c", "a");
        },"c").start();
    }
}

class SyncWaitNotify {

    private String startThread;
    private int loopNumber;

    public SyncWaitNotify(String startThread, int loopNumber) {
        this.startThread = startThread;
        this.loopNumber = loopNumber;
    }

    public void print(String threadName, String nextThread) {
        for (int i = 0; i < loopNumber; i++) {
            synchronized (this) {
                while (!threadName.equals(this.startThread)) {
                    try {
                        this.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.print(threadName);
                startThread = nextThread;
                this.notifyAll();
            }
        }
    }
}

abcabcabcabcabc

（2）park & unpark

public class SwapOrder_ParkUnpark {

    static Thread a;
    static Thread b;
    static Thread c;

    public static void main(String[] args) throws InterruptedException {
        ParkUnpark parkUnpark = new ParkUnpark(5);
        a = new Thread(() -> {
            parkUnpark.print("a", b);
        }, "a");
        b = new Thread(() -> {
            parkUnpark.print("b", c);
        }, "b");
        c = new Thread(() -> {
            parkUnpark.print("c", a);
        }, "c");

        parkUnpark.setThreads(a, b, c);
        parkUnpark.start();
        LockSupport.unpark(a);
    }
}

class ParkUnpark {

    private int loopNum;
    private Thread[] threads;

    public ParkUnpark(int loopNum) {
        this.loopNum = loopNum;
    }

    public void setThreads(Thread... threads) {
        this.threads = threads;
    }

    public void print(String content, Thread nextThread) {
        for (int i = 0; i < loopNum; i++) {
            LockSupport.park();
            System.out.print(content);
            LockSupport.unpark(nextThread);
        }
    }

    public void start() {
        for (Thread thread : threads) {
            thread.start();
        }
    }
}

abcabcabcabcabc

（3）ReentrantLock使用条件变量Condition

public class SwapOrder_ConditionReentrantLock {

    static Thread a;
    static Thread b;
    static Thread c;
    static Condition aCondition;
    static Condition bCondition;
    static Condition cCondition;

    public static void main(String[] args) throws InterruptedException {
        AwaitSignal as = new AwaitSignal(5);
        aCondition = as.newCondition();
        bCondition = as.newCondition();
        cCondition = as.newCondition();
        a = new Thread(() -> {
            as.print("a", aCondition, bCondition);
        },"a");
        b = new Thread(() -> {
            as.print("b", bCondition, cCondition);
        },"b");
        c = new Thread(() -> {
            as.print("c", cCondition, aCondition);
        },"c");

        a.start();
        b.start();
        c.start();
        Thread.sleep(1000);
        as.start(aCondition);
    }
}

class AwaitSignal extends ReentrantLock {

    public void start(Condition first) {
        this.lock();
        try {
            first.signal();
        } finally {
            this.unlock();
        }
    }

    public void print(String str, Condition current, Condition next) {
        for (int i = 0; i < loopNumber; i++) {
            this.lock();
            try {
                current.await();
                System.out.print(str);
                next.signal();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                this.unlock();
            }
        }
    }

    // 循环次数
    private int loopNumber;

    public AwaitSignal(int loopNumber) {
        this.loopNumber = loopNumber;
    }
}

abcabcabcabcabc

全面深入学习Java并发编程-第五章

Java相关栏目本月热门文章