4.1 等待/通知机制

4.1.1 概念

线程条件不满足时进入等待状态，条件满足时继续执行。

4.1.2 实现

Object类中的wait()方法可以使执行当前代码的线程等待，暂停执行，直到通知或者被中断为止。

  1. public class Test01 {
  2.     public static void main(String[] args) {
  3.         try{
  4.             String text = "something";
  5.             System.out.println("同步前的代码");
  6.             synchronized (text){
  7.                 System.out.println("同步代码块开始……");
  8.                 text.wait();    // 线程等待，后面的内容不输出
  9.                 System.out.println("wait后的代码……");
  10.             }
  11.             System.out.println("同步代码块后面的代码");
  12.         } catch (InterruptedException e) {
  13.             e.printStackTrace();
  14.         }
  15.         System.out.println("main最后的代码");
  16.     }
  17. }

注意：1. wait()方法只能在同步代码块中由锁对象调用；

  1. 调用wait()方法，当前线程会释放锁；

Object类的notify()可以唤醒线程，该方法也必须在同步代码块中，由锁对象调用，没有使用锁对象调用wait()/notify()会抛出IllgalMonitorStateException异常。如果有多个等待的线程，notify()方法只能唤醒其中一个。在同步代码块中，调用notify()方法会并不会立即释放锁对象，需要等当前同步代码块执行完后会释放锁对象，一般将nofity()方法放在同步代码块的最后。
在使用notify()的时候，可能出现通过过早的情况，需要再引入一个变量来判断是否已经唤醒过，唤醒过就不再需要等待了。

  1. public class Test02 {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         /*
  4.         线程1开始等待：1614737148963
  5.         线程2开始唤醒：1614737151964
  6.         线程2结束唤醒：1614737152964
  7.         线程1结束等待：1614737152964
  8.          */
  9.         String lock = "something";
  10.         Thread t1 = new Thread(() -> {
  11.             synchronized (lock) {
  12.                 System.out.println("线程1开始等待：" + System.currentTimeMillis());
  13.                 try {
  14.                     lock.wait();
  15.                 } catch (InterruptedException e) {
  16.                     e.printStackTrace();
  17.                 }
  18.                 System.out.println("线程1结束等待：" + System.currentTimeMillis());
  19.             }
  20.         });
  22.         Thread t2 = new Thread(() -> {
  23.             synchronized (lock) {
  24.                 System.out.println("线程2开始唤醒：" + System.currentTimeMillis());
  25.                 lock.notify();
  26.                 try {
  27.                     Thread.sleep(1000);
  28.                 } catch (InterruptedException e) {
  29.                     e.printStackTrace();
  30.                 }
  31.                 System.out.println("线程2结束唤醒：" + System.currentTimeMillis());
  32.             }
  33.         });
  35.         t1.start();
  37.         Thread.sleep(3000);
  39.         t2.start();
  40.     }
  41. }

4.1.3 interrupt()方法会中断wait()

当线程处于wait()等待状态时，调用线程对象的interrupt()方法会中断线程的等待状态，会产生InterruptedException异常。

  1. public class Test03 {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         SubThread subThread = new SubThread();
  4.         subThread.start();
  6.         Thread.sleep(2000);
  7.         subThread.interrupt();
  8.     }
  10.     private static final Object LOCK = new Object();
  12.     static class SubThread extends Thread {
  13.         @Override
  14.         public void run() {
  15.             synchronized (LOCK) {
  16.                 System.out.println("begin wait...");
  17.                 try {
  18.                     LOCK.wait();
  19.                     System.out.println("end wait...");
  20.                 } catch (InterruptedException e) {
  21.                     System.out.println("wait等待被中断");
  22.                 }
  23.             }
  24.         }
  25.     }
  26. }

4.1.4 notify()与notifyAll()

notify()一次只能唤醒一个线程，如果有多个等待线程，只能随机唤醒其中某一个；想要唤醒所有线程，需要调用notifyAll()。

  1. public class Test04 {
  2.     public static void main(String[] args) {
  3.         Object lock = new Object();
  4.         SubThread t1 = new SubThread(lock);
  5.         SubThread t2 = new SubThread(lock);
  6.         SubThread t3 = new SubThread(lock);
  7.         t1.setName("t1");
  8.         t2.setName("t2");
  9.         t3.setName("t3");
  10.         t1.start();
  11.         t2.start();
  12.         t3.start();
  14.         try {
  15.             Thread.sleep(2000);
  16.         } catch (InterruptedException e) {
  17.             e.printStackTrace();
  18.         }
  20.         synchronized (lock){
  21.             /* 只能唤醒一个线程，这种情况称为信号丢失
  22.             t1 -- begin wait...
  23.             t3 -- begin wait...
  24.             t2 -- begin wait...
  25.             t1 -- end wait...           
  26.              */
  27.             // lock.notify(); 
  28.             /*
  29.             t1 -- begin wait...
  30.             t3 -- begin wait...
  31.             t2 -- begin wait...
  32.             t2 -- end wait...
  33.             t3 -- end wait...
  34.             t1 -- end wait...
  35.              */
  36.             lock.notifyAll();
  37.         }
  39.     }
  41.     static class SubThread extends Thread {
  42.         private Object lock;
  44.         public SubThread(Object lock){
  45.             this.lock = lock;
  46.         }
  48.         @Override
  49.         public void run() {
  50.             synchronized (lock){
  51.                 try{
  52.                     System.out.println(Thread.currentThread().getName() + " -- begin wait...");
  53.                     lock.wait();
  54.                     System.out.println(Thread.currentThread().getName() + " -- end wait...");
  55.                 } catch (InterruptedException e) {
  56.                     e.printStackTrace();
  57.                 }
  58.             }
  59.         }
  60.     }
  61. }

4.1.5 wait(long)

在参数指定的时间内没有被唤醒，超时后会自动唤醒。

4.1.6 wait等待条件发生变化

  1. /**
  2.  * wait条件发生变化
  3.  * 定义一个集合
  4.  * 定义一个线程向集合中添加数据，添加完数据会通知另外的线程从集合中取数据
  5.  * 定义一个线程从集合中取数据，如果集合中没有数据就等待
  6.  */
  7. public class Test05 {
  8.     public static void main(String[] args) {
  9.         ThreadAdd threadAdd = new ThreadAdd();
  10.         ThreadSubtract threadSubtract = new ThreadSubtract();
  11.         threadSubtract.setName("subtract 1");
  13.         /*
  14.         测试一：先开启添加数据的线程，再开启一个取数据的线程，大多数情况下正常
  15.                 threadAdd.start();
  16.                 threadSubtract.start();
  17.         测试二：先开启取数据的线程，再开启添加数据线程
  18.                 threadSubtract.start();
  19.                 threadAdd.start();
  20.         测试三：先开启两个取数据的线程，再开启添加数据的线程
  21.                  ThreadSubtract threadSubtract2 = new ThreadSubtract();
  22.                  threadSubtract2.setName("subtract 2");
  23.                  threadSubtract.start();
  24.                  threadSubtract2.start();
  25.                  threadAdd.start();
  26.                结果：同时唤醒后，先后取数据，一个取到，另一个取数据时再取时出现异常
  27.                     subtract 1 begin wait...
  28.                     subtract 2 begin wait...
  29.                     subtract 2 end wait...
  30.                     subtract 2从集合中取了data 后，集合中数据的数量：0
  31.                     subtract 1 end wait...
  32.                解决方案：被唤醒后依然要再判断，可将if改为while
  33.          */
  34.     }
  36.     /**
  37.      * 1） 定义List集合
  38.      */
  39.     static List<String> list = new ArrayList<>();
  41.     /**
  42.      * 2) 定义方法从集合中取数据
  43.      */
  44.     public static void subtract() {
  45.         synchronized (list) {
  46.             try {
  47.                 if (list.size() == 0) {
  48.                     System.out.println(Thread.currentThread().getName() + " begin wait...");
  49.                     list.wait();
  50.                     System.out.println(Thread.currentThread().getName() + " end wait...");
  51.                 }
  52.             } catch (InterruptedException e) {
  53.                 e.printStackTrace();
  54.             }
  55.             Object data = list.remove(0);
  56.             System.out.println(Thread.currentThread().getName()
  57.                     + "从集合中取了" + data + " 后，集合中数据的数量：" + list.size());
  58.         }
  59.     }
  61.     /**
  62.      * 3) 定义方法向集合中添加数据后，通知等待的线程取数据
  63.      */
  64.     public static void add(){
  65.         synchronized (list){
  66.             list.add("data");
  67.             list.notifyAll();
  68.         }
  69.     }
  71.     /**
  72.      * 4) 定义线程类调用subtract取数据
  73.      */
  74.     static class ThreadSubtract extends Thread {
  75.         @Override
  76.         public void run() {
  77.             subtract();
  78.         }
  79.     }
  81.     /**
  82.      * 5) 定义线程类调用add添加数据
  83.      */
  84.     static class ThreadAdd extends Thread {
  85.         @Override
  86.         public void run() {
  87.             add();
  88.         }
  89.     }
  90. }

4.1.7 生产者消费者模式

  1. /**
  2.  * 模拟产品
  3.  */
  4. public class ValueOP {
  5.     private String value = "";
  7.     /**
  8.      * 生产过程
  9.      */
  10.     public void setValue() {
  11.         synchronized (this){
  12.             while (!"".equalsIgnoreCase(value)){
  13.                 try {
  14.                     this.wait();
  15.                 } catch (InterruptedException e) {
  16.                     e.printStackTrace();
  17.                 }
  18.             }
  19.             String value = System.currentTimeMillis() + " - " + System.nanoTime();
  20.             System.out.println("set设置的值是：" + value);
  21.             this.value = value;
  22.             this.notifyAll();
  23.         }
  24.     }
  26.     /**
  27.      * 消费过程
  28.      */
  29.     public void getValue(){
  30.         synchronized (this){
  31.             while ("".equalsIgnoreCase(value)){
  32.                 try {
  33.                     this.wait();
  34.                 } catch (InterruptedException e) {
  35.                     e.printStackTrace();
  36.                 }
  37.             }
  38.             System.out.println("get的值是：" + this.value);
  39.             this.value = "";
  40.             this.notifyAll();
  41.         }
  42.     }
  43. }
  45. /**
  46.  * 定义线程模拟生产者
  47.  */
  48. public class ProducerThread extends Thread{
  49.     /**
  50.      * 生产者生产数据，调用valueOP类的setValue方法给value赋值
  51.      */
  52.     private ValueOP obj;
  54.     public ProducerThread(ValueOP obj){
  55.         this.obj = obj;
  56.     }
  58.     @Override
  59.     public void run() {
  60.         while (true){
  61.             obj.setValue();
  62.         }
  63.     }
  64. }
  66. /**
  67.  * 定义线程模拟消费者
  68.  */
  69. public class ConsumerThread extends Thread{
  70.     /**
  71.      * 消费者消费数据，调用valueOP类的getValue方法消费产品
  72.      */
  73.     private ValueOP obj;
  75.     public ConsumerThread(ValueOP obj){
  76.         this.obj = obj;
  77.     }
  79.     @Override
  80.     public void run() {
  81.         while (true){
  82.             obj.getValue();
  83.         }
  84.     }
  85. }
  87. public class Test {
  88.     public static void main(String[] args) {
  89.         ValueOP valueOP = new ValueOP();
  90.         /*
  91.          测试一生产，一消费的情况
  92.             ProducerThread producerThread = new ProducerThread(valueOP);
  93.             ConsumerThread consumerThread = new ConsumerThread(valueOP);
  95.             producerThread.start();
  96.             consumerThread.start();
  97.          测试多个生产者，多个消费者：
  98.             把等待条件的if改为while，避免重新唤醒消费者后直接消费空串；
  99.             但是出现了假死现象：消费者消费空串时，重新进入等待状态，生产者又未被唤醒进行生产，所以需要把notify()改为notifyAll()。
  100.             ProducerThread producerThread1 = new ProducerThread(valueOP);
  101.             ProducerThread producerThread2 = new ProducerThread(valueOP);
  102.             ProducerThread producerThread3 = new ProducerThread(valueOP);
  103.             ConsumerThread consumerThread1 = new ConsumerThread(valueOP);
  104.             ConsumerThread consumerThread2 = new ConsumerThread(valueOP);
  105.             ConsumerThread consumerThread3 = new ConsumerThread(valueOP);
  106.             producerThread1.start();
  107.             producerThread2.start();
  108.             producerThread3.start();
  109.             consumerThread1.start();
  110.             consumerThread2.start();
  111.             consumerThread3.start();
  112.          */
  113.     }
  114. }

4.2 通过管道实现线程间的通信

在java.io包中的PipeStream管道流用于在线程之间传送数据，一个线程发送数据到输出管道，另一个线程从输入管道中读取数据，相关的类包括：PipedInputStream和PipedOutputStream，PipedReader和PipedWriter。

  1. /**
  2.  * 使用PipedInputStream和PipedOutputStream管道字节流在线程之间传递数据
  3.  *
  4.  * @author 王游
  5.  * @date 2021/3/3 19:45
  6.  */
  7. public class Test {
  8.     public static void main(String[] args) {
  9.         // 定义管道字节流
  10.         PipedInputStream inputStream = new PipedInputStream();
  11.         PipedOutputStream outputStream = new PipedOutputStream();
  12.         // 建立连接
  13.         try {
  14.             inputStream.connect(outputStream);
  15.         } catch (IOException e) {
  16.             e.printStackTrace();
  17.         }
  19.         new Thread(() -> readData(inputStream)).start();
  20.         try {
  21.             Thread.sleep(1000);
  22.         } catch (InterruptedException e) {
  23.             e.printStackTrace();
  24.         }
  25.         new Thread(() -> writeData(outputStream)).start();
  26.     }
  28.     /**
  29.      * 定义方法向管道流中写入数据
  30.      */
  31.     public static void writeData(PipedOutputStream out) {
  32.         int sum = 100;
  33.         try {
  34.             for (int i = 0; i < sum; i++) {
  35.                 String data = "" + i;
  36.                 out.write(data.getBytes());
  37.             }
  38.             out.close();
  39.         } catch (IOException e) {
  40.             e.printStackTrace();
  41.         }
  42.     }
  44.     /**
  45.      * 定义方法从管道流中读取数据
  46.      */
  47.     public static void readData(PipedInputStream in) {
  48.         byte[] bytes = new byte[1024];
  49.         // 从管道中读取字节
  50.         try {
  51.             // This method blocks until input data is available, end of file is detected, or an exception is thrown.
  52.             int len = in.read(bytes);
  53.             while (len != -1) {
  54.                 System.out.println(new String(bytes, 0, len));
  55.                 len = in.read(bytes);
  56.             }
  57.             in.close();
  58.         } catch (IOException e) {
  59.             e.printStackTrace();
  60.         }
  61.     }
  62. }

4.3 thread.join

线程没有执行完之前，会一直阻塞在join方法处。

  1. public class JoinDemo extends Thread{
  2.     int i;
  3.     Thread previousThread; //上一个线程
  4.     public JoinDemo(Thread previousThread,int i){
  5.         this.previousThread=previousThread;
  6.         this.i=i;
  7.     }
  8.     @Override
  9.     public void run() {
  10.         try {
  11.           //调用上一个线程的join方法
  12.             previousThread.join(); 
  13.         } catch (InterruptedException e) {
  14.             e.printStackTrace();
  15.         }
  16.         System.out.println("num:"+i);
  17.     }
  18.     public static void main(String[] args) {
  19.         Thread previousThread=Thread.currentThread();
  20.         for(int i=0;i<10;i++){
  21.             JoinDemo joinDemo=new JoinDemo(previousThread,i);
  22.             joinDemo.start();
  23.             previousThread=joinDemo;
  24.         }
  25.     }
  26. }

注意 previousThread.join部分，在没有加join的时候运行的结果是不确定的。加了join以后，运行结果按照递增的顺序展示出来。

4.4 ThreadLocal的使用

除了控制资源的访问外，还可以通过增加资源来保证线程安全。ThreadLocal主要解决的方法是为每个线程绑定自己的值。

  1. public class Test01 {
  2.     static ThreadLocal threadLocal = new ThreadLocal();
  4.     static class SubThread extends Thread {
  5.         @Override
  6.         public void run() {
  7.             int num = 5;
  8.             for (int i = 0; i < num; i++){
  9.                 // 设置线程关联的值
  10.                 threadLocal.set(Thread.currentThread().getName() + "-" + i);
  11.                 // 调用get()方法读取关联的值
  12.                 System.out.println(Thread.currentThread().getName() + "value = " + threadLocal.get());
  13.             }
  14.         }
  15.     }
  17.     public static void main(String[] args) {
  18.         SubThread t1 = new SubThread();
  19.         SubThread t2 = new SubThread();
  21.         t1.start();
  22.         t2.start();
  23.         /*
  24.         两个线程互不干扰
  25.         Thread-0value = Thread-0-0
  26.         Thread-1value = Thread-1-1
  27.         Thread-1value = Thread-1-2
  28.         Thread-1value = Thread-1-3
  29.         Thread-1value = Thread-1-4
  30.         Thread-0value = Thread-0-1
  31.         Thread-0value = Thread-0-2
  32.         Thread-0value = Thread-0-3
  33.         Thread-0value = Thread-0-4
  34.          */
  35.     }
  36. }