多线程与并发（三）：notify与wait原理

wait notify 原理

Monitor.jpg

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

1. API介绍

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

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

public class WaitNotifyTest {

    static final Object obj = new Object();
    
    public static void main(String[] args) {

        new Thread(()->{
            synchronized (obj) {
                System.out.println("ENTER " + Thread.currentThread().getName());
                try {
                    obj.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("thread is " + Thread.currentThread().getName());
        },"t1").start();

        new Thread(()->{
            synchronized (obj) {
                System.out.println("ENTER " + Thread.currentThread().getName());
                try {
                    obj.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("thread is " + Thread.currentThread().getName());
        },"t2").start();

        new Thread(()->{

            synchronized (obj) {
                System.out.println("ENTER NOTIFY ");
                obj.notify();
            }
            System.out.println("thread is " + Thread.currentThread().getName());
        },"t3").start();

    }
}

• 测试结果：
  ENTER t1
  ENTER t2
  ENTER NOTIFY //唤醒等待线程之后，本线程需要执行完释放锁其他线程才能获得锁
  thread is t3
  thread is t1

---

2. wait 和notify 使用

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

• sleep 是Thread的方法；wait notify是Object 的方法
• sleep 不需要强制和synchronized使用；wait notify需要和synchronized 一起使用
• sleep 在睡眠的同时，不会释放对象锁；wait 在等候时会释放对象锁
• 它们的状态都是TIMED_WAITING

例子一

public class SleepTest {
    static final Object room = new Object();
    static boolean hasCigarette = false;
    static boolean hasTakeout;

    public static void main(String[] args) {

        new Thread(()->{
            synchronized (room) {
                System.out.println(Thread.currentThread().getName()+"有没有烟？" + hasCigarette);

                if(!hasCigarette){
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName()+"再问一遍有没有烟？" + hasCigarette);
                if (hasCigarette){
                    System.out.println( Thread.currentThread().getName()+"可以开始干活了");
                }
            }

        },"小男").start();

        for (int i = 0; i <5 ; i++) {
            new Thread(()->{
                synchronized (room){
                    System.out.println(Thread.currentThread().getName() + "可以开始干活了");
                }
            },"其他人" + i).start();
        }


        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        new Thread(()->{
            //synchronized (room){

            hasCigarette = true;
            System.out.println(Thread.currentThread().getName() + "送烟");
            //}
        },"小女").start();

    }
}

• 测试结果
  小男有没有烟？false
  小女送烟
  小男再问一遍有没有烟？true
  小男可以开始干活了
  其他人4可以开始干活了
  其他人3可以开始干活了
  其他人0可以开始干活了
  其他人2可以开始干活了
  其他人1可以开始干活了

---

• 其他干活的线程，都要一直阻塞，效率太低。
• 小男线程必须睡足两秒之后才能醒来，就算烟提前到，也不能立刻醒来。
• 加了synchronized(room)就好比小男在里面反锁了门睡觉，烟根本没法送进门。
• 解决方法，使用wait - notify改进

例子二

public class WaitNotifyTest1 {
    static final Object room = new Object();
    static boolean hasCigarette = false;
    static boolean hasTakeout;

    public static void main(String[] args) {

        new Thread(()->{
            synchronized (room) {
                System.out.println(Thread.currentThread().getName()+"有没有烟？" + hasCigarette);

                if(!hasCigarette){
                    try {
                      room.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName()+"再问一遍有没有烟？" + hasCigarette);
                if (hasCigarette){
                    System.out.println( Thread.currentThread().getName()+"可以开始干活了");
                }
            }

        },"小男").start();

        for (int i = 0; i <5 ; i++) {
            new Thread(()->{
                synchronized (room){
                    System.out.println(Thread.currentThread().getName() + "可以开始干活了");
                }
            },"其他人" + i).start();
        }

        new Thread(()->{
            synchronized (room){

            hasCigarette = true;
            room.notify();
            System.out.println(Thread.currentThread().getName() + "送烟");
            }
        },"小女").start();
    }
}

• 测试结果：
  小男有没有烟？false
  其他人0可以开始干活了
  其他人1可以开始干活了
  其他人2可以开始干活了
  其他人3可以开始干活了
  其他人4可以开始干活了
  小女送烟
  小男再问一遍有没有烟？true
  小男可以开始干活了

---

• 假如有其他线程也在wait,notify之后没有唤醒小男线程。

例子三

public class WaitNotifyTest2 {

    static final Object room = new Object();
    static boolean hasCigarette = false;
    static boolean hasTakeout;

    public static void main(String[] args) {

        new Thread(() -> {
            synchronized (room) {
                System.out.println(Thread.currentThread().getName() + "有没有烟？" + hasCigarette);

                if (!hasCigarette) {
                    try {
                        room.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName() + "再问一遍有没有烟？" + hasCigarette);
                if (hasCigarette) {
                    System.out.println(Thread.currentThread().getName() + "可以开始干活了");
                }
            }

        }, "小男").start();


        new Thread(() -> {
            synchronized (room) {
                System.out.println(Thread.currentThread().getName() + "外卖送到没？" + hasTakeout);
                if (!hasTakeout) {
                    try {
                        room.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName() + "再问一遍有没有送到？" + hasCigarette);
                if (hasTakeout) {
                    System.out.println(Thread.currentThread().getName() + "可以开始干活了");
                }
            }
        }, "其他人" ).start();

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(() -> {
            synchronized (room) {

                //hasCigarette = true;
                hasTakeout = true;
                room.notify();
                //room.notifyAll(); //唤醒所有等待的线程
                System.out.println(Thread.currentThread().getName() + "送到了");
            }
        }, "外卖").start();
    }
}

• 测试结果：
  小男有没有烟？false
  其他人外卖送到没？false
  外卖送到了
  小男再问一遍有没有烟？false

---

异步模式之生产者与消费者

要点

• 消费队列可以用来平衡生产和消费的线程资源
• 生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
• 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据

• JDK 中各种阻塞队列，采用的就是这种模式

  
  生产者与消费者.jpg

Park & Unpark

1. 基本使用

LockSupport 类中的方法

//暂停当前线程
LockSupport.park()；

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

示例：

public class Test2 {

    public static void main(String[] args) {
        Thread t1 = new Thread(()->{
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("start park");
            LockSupport.park();
            System.out.println("restart");
        },"t1");

        t1.start();

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("start unpark");
        LockSupport.unpark(t1);

    }
}

测试结果：
start park
start unpark
restart

---

注意两点：

1. park 之后线程状态是什么 WAITING
2. unpark 既可以在线程之前调用也可以在线程之后调用，区别是什么？
   下面原理部分说明

特点：
与Object 的wait 和notify相比

• wait notify 和notifyAll 必须配合 Object Monitor 一起使用，unpark 不必
• park unpark 是以线程为单位来 阻塞 和 唤醒线程，而notify 只能随机唤醒一个等待线程
  notifyAll 是唤醒所有等待线程
• park unpark可以先unpark ，而wait notify 不能先notify

2. park unpark 原理

每个线程都有自己的一个Parker 对象，由三部分组成_counter, _cond,_mutex 打个比喻

• 线程就像一个旅人，Parker就像他随身携带的背包，条件变量就好比背包中的帐篷。_counter就好比背包中的备用干粮（0为耗尽，1为充足）
• 调用park 就要看需不需要停下来休息
  • 如果备用干粮耗尽，那么钻进帐篷休息
  • 如果备用干粮充足，那么不需要停留，继续前进
• 调用unpark 就好比干粮充足
  • 如果这时线程还在帐篷，就唤醒让他继续前进
  • 如果这时线程还在运行，那么下次他调用park 时，近视消耗掉备用干粮，不需停留继续前进

    • 因为背包空间有限，多次调用unpark只会补充一份干粮

      
      park.jpg

1. 当前线程调用Unsafe.park()方法
2. 检查_counter,这个值为0，这时，获得_mutex互斥锁
3. 线程进入_cond 条件变量阻塞
4. 设置_counter = 0;

unpark.jpg

1. 调用Unsafe.unpark(Thread-0)方法，设置_counter为1
2. 唤醒_cond条件变量中的Thread-0
3. Thread-0 恢复运行
4. 设置_counter =0

重新理解线程状态转换

线程状态转换.jpg

假设有线程Thread t

情况一： NEW --> RUNNABLE

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

情况二：RUNNABLE <-->WAITING

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

• 调用obj.wait()方法时，t线程从RUNNABLE-->WAITING
• 调用obj.notify()，obj.notifyAll(), t.interrupt()时
  • 竞争锁成功，t 线程从WAITING --> RUNNABLE
  • 竞争锁失败，t 线程从WAITING --> BLOCKED

情况三 ：RUNNABLE <--> WAITING

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

情况四：RUNNABLE <--> WAITING

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

情况五：RUNNABLE <-->TIMED_WAITING

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

• 调用obj.wait(long n)方法时，t 线程从RUNNABLE-->TIMED_WAITING
• t 线程等待时间超过了n毫秒，或调用obj.notify() ,obj.notifyAll() , t.interrupt()时，
  • 竞争锁成功，t线程从TIMED_WAITING -->RUNNABLE
  • 竞争锁失败，t 线程从TIMED_WAITING --> BLOCKED

情况六: RUNNABLE <--> TIMED_WAITING

• 当前线程调用t.join(long n)方法时，当前线程从RUNNABLE --> TIMED_WAITING
  • 注意，当前线程是在 t线程对象的监视器上等待
• 当前线程等待时间超过n毫秒，或t线程运行结束，或调用了当前线程的interrupt()时，当前线程从TIME_WAITING --> RUNNABLE

情况七： RUNNABLE <--> TIMED_WAITING

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

情况八： RUNNABLE <--> TIMED_WAITING

• 当前线程调用LockSupport.parkNanos(long n)或LockSupport.parkUntil(long n)时，当前线程从RUNNABLE --> TIMED_WAITING
• 调用LockSupport.unpark(目标线程)或调用了线程的interrupt()，或是等待超时，会让目标线程从TIMED_WAITING --> RUNNABLE

情况九： RUNNABLE <--> BLOCKED

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

情况十： RUNNABLE <--> TERMINATED

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