03 Java的wait/notify及其应用

3 Java中wait/notify

3-1 原理

如上图所示：

step1：线程1之前获得过Monitor，在执行临界区代码时发现部分条件不满足，无法执行完代码，因此主动调用wait让出坑位，自己进入WaitSet ，让其他阻塞的线程能够获得Monitor,避免浪费资源。

step2: 线程1主动放弃Monitor，会唤醒BLOCKED的线程去获得Monitor，图中线程2获得了Monitor。

• 如果条件满足，此时线程2可以主动调用 notify 或 notifyAll 去唤醒WAITING的线程，唤醒后并不意味者立刻获得锁，仍需进入EntryList 重新竞争。

3-2 wait与notify API的使用

obj.wait()      让进入 object 监视器的线程到 waitSet 等待
obj.wait(n)     让进入 object 监视器的线程到 waitSet 等待一定时间然后唤醒
obj.notify()    在 object 上正在 waitSet 等待的线程中随机挑一个唤醒
obj.notifyAll() 让 object 上正在 waitSet 等待的线程全部唤醒

注意:

• 都是线程之间进行协作的手段，都属于 Object 对象的方法。（Java中所有类都是Object的子类）不是Thread类专属方法。
• 必须获得此对象的锁，才能调用这几个方法。否则会产生异常（如下所示）。

package chapter3;
import lombok.extern.slf4j.Slf4j;
@Slf4j(topic = "c.Test6")
public class Test6 {
    public static void main(String[] args) throws InterruptedException {
        String tmp = "";
        tmp.wait();
    }
}

运行结果：

Exception in thread "main" java.lang.IllegalMonitorStateException
    at java.lang.Object.wait(Native Method)
	at java.lang.Object.wait(Object.java:502)
	at chapter3.Test6.main(Test6.java:8)

使用wait与notify的简单的实例

package chapter3;
import lombok.extern.slf4j.Slf4j;
@Slf4j(topic = "c.Test6")
public class Test6 {
    public static void main(String[] args) throws InterruptedException {
        String tmp = "";
        new Thread(()->{
            synchronized (tmp){
                log.warn("this is thread 1");
                try {
                    tmp.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.warn("thread 1 run after waiting.");
            }
        },"t1").start();
        
        new Thread(()->{
            synchronized (tmp){
                log.warn("this is thread 2");
                try {
                    tmp.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.warn("thread 2 run after waiting.");
            }
        },"t2").start();
        Thread.sleep(5000);            // 主线程休眠一会
        // 必须成为monitor的ownerh后(获得锁)，才有资格wait,notify
        log.warn("start notify threads.");
        synchronized (tmp){
            tmp.notify();             // 主线程任意唤醒waitset中的一个线程
//            tmp.notifyAll();        // 主线程任意唤醒waitset中的所有线程
        }
    }
}

运行结果：

[t1] WARN c.Test6 - this is thread 1
[t2] WARN c.Test6 - this is thread 2
[main] WARN c.Test6 - start notify threads.
[t1] WARN c.Test6 - thread 1 run after waiting.

3-3 wait与sleep的区别（等待与睡眠的区别）

sleep(n)与wait(n)：

• 不同点
  • sleep是Thread类的静态方法，而wait是所有对象的方法。
  • wait的使用必须配合synchronized,并且会释放对象锁，而在synchroized代码块中sleep，锁不会被释放。这一点是可以用于区分它们的使用场景。
• 共同点：
  • 使用后，线程都会进入JAVA API层面的TIME WAITING状态

3-4 wait与notify的使用模板

虚假唤醒：当多个线程由于执行条件不满足使用wait进入 WAITING状态。此时只有部分线程的条件满足却notify所有线程。对于那些执行条件仍未满足的线程来说就是虚假唤醒。

• 因此在代码编写时，当线程被虚假唤醒后仍然需要判断条件是否满足，不满足则继续wait。

避免虚假唤醒引发问题的模板如下：

synchronized(lock) {
	while(条件不成立) {       //while语句保证线程执行的条件必须满足，避免条件不满足的情况下执行程序
    	lock.wait();
    }
    // 干活
}

//另一个线程
synchronized(lock) {
    lock.notifyAll();
}

4 同步模式-保护性暂停（wait与notify的应用）

4-1 基础

定义：保护性暂停（Guarded Suspension），针对一个线程等待另外一个线程的场景。

知识点：

• 传递结果的2个线程之间都关联同一个GuardedObject
• JDK 中，join 的实现、Future 的实现，采用的就是此模式
• 保护性暂停是同步模式

保护性暂停的一个实例

package chapter3;
import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

class GuardedObject{
    private Object response = null;
    public Object get(){
        synchronized (this){
            while(response == null){
                try{
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        return response;
    }

    public void complete(Object response){
        synchronized (this){
            this.response = response;
            this.notifyAll();
        }
    }

}

@Slf4j(topic = "c.Test7")
class Test7{
    public static void main(String[] args) {
        log.warn("start the main thread!");
        GuardedObject guard = new GuardedObject();

        // 定义一个线程t1去获取另外一个线程t2网页下载的结果
        new Thread(()->{
            log.warn("等待网页下载");
            List<String> downres = (List<String>) guard.get();
            log.warn("已经获得下载的网页");
            log.warn("网页的大小{}",downres.size());
        },"t1").start();

        new Thread(()->{
            List<String> tmp;
            try {
                tmp = Downloader.download();
                guard.complete(tmp);
                log.warn("下载完成");
            } catch (IOException e) {
                e.printStackTrace();
            }
        },"t2").start();
    }
}

class Downloader{
    public  static List<String> download() throws IOException {
        HttpURLConnection conn = (HttpURLConnection) new URL("https://www.baidu.com/").openConnection();
        List<String> lines = new ArrayList<>();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))
        ){
                String tmp;
                while((tmp = reader.readLine()) != null){
                    lines.add(tmp);
                }

            };
        return lines;
    }
}

运行结果：

[main] WARN c.Test7 - start the main thread!
[t1] WARN c.Test7 - 等待网页下载
[t2] WARN c.Test7 - 下载完成
[t1] WARN c.Test7 - 已经获得下载的网页
[t1] WARN c.Test7 - 网页的大小3

总结：上面的代码中线程t1等待线程t2准备好对象，并通过GuardedObject获得对象。

优势：

• 在使用join实现2个线程的同步时，必须把传递的对象设为公共变量。

有时间限制的保护性暂停的实例

package chapter3;

import javafx.beans.binding.ObjectExpression;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

// 定义了一个guard  object进行有时限的等待
class GuardedObjectTime{
    private Object response = null;
    // timeout:线程最多等待的时间
    public Object get(long timeout){
        synchronized (this){
            long begin = System.currentTimeMillis();
            long passedTime = 0;
            while(response == null){
                // 之所以单独弄一个waittime是因为需要考虑虚假唤醒的情况
                // 虚假唤醒后的线程之前等待的时间也要从总的等待时间减去
                long waittime = timeout - passedTime;
                if(waittime <= 0)
                    break;
                try{
                    this.wait(waittime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                passedTime = System.currentTimeMillis()-begin;
            }
        }
        return response;
    }

    public void complete(Object response){
        synchronized (this){
            this.response = response;
            this.notifyAll();
        }
    }

}

@Slf4j(topic = "c.Test7")
class Test8{
    public static void main(String[] args) {
        log.warn("start the main thread!");
        GuardedObjectTime guard = new GuardedObjectTime();

        // 定义一个线程t1去获取另外一个线程t2的结果
        new Thread(()->{
            Object response = guard.get(2000);      // 获取结果最多等待2s
            log.warn("The result is {}",response);
        },"t1").start();

        new Thread(()->{
            Object tmp = new Object();
            try {
//                Thread.sleep(3000);            // 线程等待时间超过规定时间，无法得到任何结果
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            guard.complete(tmp);
        },"t2").start();
    }
}

线程t2睡眠时间为3000ms的运行结果：

[main] WARN c.Test7 - start the main thread!
[t1] WARN c.Test7 - The result is null

线程t2睡眠时间为1000ms的运行结果：

[main] WARN c.Test7 - start the main thread!
[t1] WARN c.Test7 - The result is java.lang.Object@54ed57cd

4-2 Java的join的原理(底层使用也是wait)

join方法的源代码如下：

    /**
     * Waits for this thread to die.
     *
     * <p> An invocation of this method behaves in exactly the same
     * way as the invocation
     *
     * <blockquote>
     * {@linkplain #join(long) join}{@code (0)}
     * </blockquote>
     *
     * @throws  InterruptedException
     *          if any thread has interrupted the current thread. The
     *          <i>interrupted status</i> of the current thread is
     *          cleared when this exception is thrown.
     */
     // 可以看到没有参数的join调用的是带时间的join
    public final void join() throws InterruptedException {
        join(0);
    }
    
public final synchronized void join(long millis)
    throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;
        // 情况1：等待时间小于0
        if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
        }
        // 情况2:等待时间等于0
        if (millis == 0) {
            while (isAlive()) {
                wait(0);
            }
        } else {
        // ============================================================================
        // 情况3：等待时间>0, 可以看到这段代码的逻辑 *有时间限制的保护性暂停的实例*中get的方法逻辑相一致
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                wait(delay);
                now = System.currentTimeMillis() - base;
            }
        // =============================================================================
        }
    }
    
    

总结：join的源码的实现也是利用了保护性暂停的设计模式。

4-3 同步模式之保护性暂停扩展(多个线程使用多个GuardObject的解耦方法)

场景：考虑邮递员，收件人，信箱三种线程对象：

• 多名邮递员(产生结果的线程)：每个邮递员发送邮件
• 信箱（多个GuardObject）
• 多名收件人(接受结果线程)

动机：多个类之间使用GuardedObject对象作为参数传递不是很方便，因此需要设计一个解耦的中间类，能够做到以下几点：

• 支持多个任务的管理
• 解耦“结果等待者”和“结果生产者”。

实例：（自己写的test9代码有问题，待解决）

基本思想：

step1: 为guardedObject添加id。

step2: 定义管理类管理多个guardedObject,有以下功能：

• 创建guardedObject实例并产生唯一id并放入容器管理
• 根据id返回对应的guardedObject对象

step3:业务类调用管理类实现线程之间的通信

• 生产者利用管理类创建其对应的guardedObject实例用于接收信息
• 消费者利用管理类获取对应的guardedObject实例用于发送信息。

注意点：

• 下面代码中 Mailboxes 是通用类，用于管理多个guardedObject对象，可以复用。
• 下面代码中Postman与People是业务相关的类。

package chapter4;

import java.util.Hashtable;
import java.util.Map;
import java.util.Set;
import chapter2.Sleeper;
import lombok.extern.slf4j.Slf4j;

class GuardedObject {
    private int id;
    public GuardedObject(int id) {
        this.id = id;
    }
    public int getId() {
        return id;
    }
    private Object response;
    public Object get(long timeout) {
        synchronized (this) {
            long begin = System.currentTimeMillis();
            long passedTime = 0;
            while (response == null) {
                long waitTime = timeout - passedTime;
                if (timeout - passedTime <= 0) {
                    break;
                }
                try {
                    this.wait(waitTime); // 虚假唤醒 15:00:01
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                passedTime = System.currentTimeMillis() - begin; // 15:00:02 1s
            }
            return response;
        }
    }
    public void complete(Object response) {
        synchronized (this) {
            this.response = response;
            this.notifyAll();
        }
    }
}

class Mailboxes {
    private static Map<Integer, GuardedObject> boxes = new Hashtable<>();
    private static int id = 1;
    // 产生唯一 id
    private static synchronized int generateId() {
        return id++;
    }
    // 由于HashTABLE是线程安全的，所以下面2个方法不需要去synchroized
    public static GuardedObject getGuardedObject(int id) {
        return boxes.remove(id);
    }
    public static GuardedObject createGuardedObject() {
        GuardedObject go = new GuardedObject(generateId());
        boxes.put(go.getId(), go);
        return go;
    }
    public static Set<Integer> getIds() {
        return boxes.keySet();
    }
}

@Slf4j(topic = "c.People")
class People extends Thread{
    @Override
    public void run() {
        // 收信
        GuardedObject guardedObject = Mailboxes.createGuardedObject();
        log.warn("开始收信 id:{}", guardedObject.getId());
        Object mail = guardedObject.get(5000);
        log.warn("收到信 id:{}, 内容:{}", guardedObject.getId(), mail);
    }
}

@Slf4j(topic = "c.PostMan")
class Postman extends Thread {
    private int id;
    private String mail;
    public Postman(int id, String mail) {
        this.id = id;
        this.mail = mail;
    }
    @Override
    public void run() {
        GuardedObject guardedObject = Mailboxes.getGuardedObject(id);
        log.warn("送信 id:{}, 内容:{}", id, mail);
        guardedObject.complete(mail);
    }
}

@Slf4j(topic = "c.test1")
public class test1 {
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 3; i++) {
            new People().start();
        }
        System.out.println(Mailboxes.getIds());
        Thread.sleep(1000);
        System.out.println(Mailboxes.getIds());
        for (Integer id : Mailboxes.getIds()) {
            new Postman(id, "内容" + id).start();
        }
    }
}

运行结果

[]
[Thread-2] WARN c.People - 开始收信 id:3
[Thread-0] WARN c.People - 开始收信 id:2
[Thread-1] WARN c.People - 开始收信 id:1
[3, 2, 1]
[Thread-3] WARN c.PostMan - 送信 id:3, 内容:内容3
[Thread-4] WARN c.PostMan - 送信 id:2, 内容:内容2
[Thread-2] WARN c.People - 收到信 id:3, 内容:内容3
[Thread-0] WARN c.People - 收到信 id:2, 内容:内容2
[Thread-5] WARN c.PostMan - 送信 id:1, 内容:内容1
[Thread-1] WARN c.People - 收到信 id:1, 内容:内容1

5 异步模式-生产者与消费者模式（wait/notify应用）

5-1 概述

生产者与消费者模式特点：

• 该模式使用消息队列来平衡生产和消费的线程资源
• 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据

应用场景：JDK 中各种阻塞队列，采用的就是这种模式

问题：生产者与消费者模式与保护性暂停模式的区别？

• 保护性暂停模式必须是一对一的，线程发送的信息可以被其对应的接受线程立刻接受到，所以是同步模式
• 生产者与消费者模式没有一对一的限制，此外需要消息对列传递信息，信息的传递存在延迟，所以是异步模式。

5-2 代码实践

package chapter4;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.*;
import lombok.extern.slf4j.Slf4j;
// 获取网页下载的内容，并存入到List
class Downloader{
    public static List<String> download() throws IOException {
        HttpURLConnection conn = (HttpURLConnection) new URL("https://www.baidu.com/").openConnection();
        List<String> lines = new ArrayList<>();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))
        ){
            String tmp;
            while((tmp = reader.readLine()) != null){
                lines.add(tmp);
            }

        };
        return lines;
    }
}

final class Message {
    private int id;
    private Object message;

    public Message(int id, Object message) {
        this.id = id;
        this.message = message;
    }

    public int getId() {
        return id;
    }

    public Object getMessage() {
        return message;
    }
}

@Slf4j(topic = "c.MessageQueue")
class MessageQueue {
    // 使用双向链表实现消息队列，用于存储message instance
    private LinkedList<Message> queue;
    private int capacity;
    public MessageQueue(int capacity) {
        this.capacity = capacity;
        queue = new LinkedList<>();
    }
    public Message take() {
        synchronized (queue) {
            // 消费者线程判断消息队列是否为空，为空则等待
            while (queue.isEmpty()) {
                log.warn("没货了, wait");
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 从队列的头部获取message
            Message message = queue.removeFirst();
            queue.notifyAll();
            return message;
        }
    }
    public void put(Message message) {
        synchronized (queue) {
            // 生产者线程判断队列是否已满，为空则等待
            while (queue.size() == capacity) {
                log.warn("库存已达上限, wait");
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(message);
            queue.notifyAll();
        }
    }
}

@Slf4j(topic = "c.test2")
public class test2 {
    public static void main(String[] args) {
        MessageQueue messageQueue = new MessageQueue(2);
        // 4 个生产者线程, 下载任务
        for (int i = 0; i < 4; i++) {
            int id = i;
            new Thread(() -> {
                try{
                    List<String> response = Downloader.download();
                    log.warn("try put message({})", id);
                    messageQueue.put(new Message(id, response));
                 } catch (IOException e) {
                    e.printStackTrace();
                 }

            }, "生产者" + i).start();
        }
        new Thread(() -> {
            while (true) {
                Message message = messageQueue.take();
                List<String> response = (List<String>) message.getMessage();
                log.warn("take message({}): [{}] lines", message.getId(), response.size());
            }
        }, "消费者").start();
    }
}

总结：生产者与消费者模式本质上依旧是wait/notify组合的应用。

• 上述代码中采用双端队列arrayList充当消息队列，并限制了消息队列的大小。
• 生产者
  • 消息队列未满的情况下可以放入东西，通过notify通知消费者“东西放好了，继续拿”
  • 消息队列满的情况下wait
• 消费者
  • 消息队列未空的情况下可以拿走东西，通过notify通知生产者“东西拿走了，继续放”
  • 消费队列为空，则wait

参考资料

JAVA中的util参考手册

并发编程课程

20210228

03 Java的wait/notify及其应用

原文：https://www.cnblogs.com/kfcuj/p/14458395.html