多线程|wait和notify实现线程之间的通信 java-ee|阻塞队列|生产者消费者

目录
一. 为什么需要线程通信
二. wait和notify方法
【多线程|wait和notify实现线程之间的通信】1. wait()方法
2. notify()方法
3. notifyAll()方法
三. 使用wait和notify实现面包房业务
四. 阻塞队列
1. 生产者消费者模型
2. 标准库中的阻塞队列
3. 阻塞队列的模拟实现

一. 为什么需要线程通信

线程是并发并行的执行，表现出来是线程随机执行，但是我们在实际应用中对线程的执行顺序是有要求的，这就需要用到线程通信

线程通信为什么不使用优先级来来解决线程的运行顺序？

总的优先级是由线程pcb中的优先级信息和线程等待时间共同决定的，所以一般开发中不会依赖优先级来表示线程的执行顺序

看下面这样的一个场景：面包房的例子来描述生产者消费者模型

有一个面包房，里面有面包师傅和顾客，对应我们的生产者和消费者，而面包房有一个库存用来存储面包，当库存满了之后就不在生产，同时消费者也在购买面包，当库存面包卖完了之后，消费者必须等待新的面包生产出来才能继续购买

分析：对于何时停止生产何时停止消费就需要应用到线程通信来准确的传达生产和消费信息

二. wait和notify方法

wait()：让当前线程持有的对象锁释放并等待
wait(long timeout)：对应的参数是线程等待的时间
notify()：唤醒使用同一个对象调用wait进入等待的线程，重新竞争对象锁
notifyAll()：如果有多个线程等待，notifyAll是全部唤醒，notify是随机唤醒一个

??注意：

这几个方法都属于Object类中的方法
必须使用在synchronized同步代码块/同步方法中
哪个对象加锁，就是用哪个对象wait，notify
调用notify后不是立即唤醒，而是等synchronized结束以后，才唤醒

1. wait()方法调用wait方法后：

使执行当前代码的线程进行等待（线程放在等待队列）
释放当前的锁
满足一定条件时被唤醒，重新尝试获取锁

wait等待结束的条件：

其他线程调用该对象的notify方法
wait等待时间超时（timeout参数来指定等待时间）
其他线程调用interrupted方法，导致wait抛出InterruptedException异常

2. notify()方法

当使用wait不带参数的方法时，唤醒线程等待就需要使用notify方法

这个方法是唤醒那些等待该对象的对象锁的线程，使他们可以重新获取该对象的对象锁
如果有多个线程等待，则由线程调度器随机挑选出一个呈wait 状态的线程(不存在先来后到)
在notify()方法后，当前线程不会马上释放该对象锁，要等到执行notify()方法的线程将程序执行完，也就是退出同步代码块之后才会释放对象锁

3. notifyAll()方法

该方法和notify()方法作用一样，只是唤醒的时候，将所有等待的线程都唤醒
notify()方法只是随机唤醒一个线程

三. 使用wait和notify实现面包房业务前提说明：

有2个面包师傅，面包师傅一次可以做出两个面包
仓库可以存储100个面包
有10个消费者，每个消费者一次购买一个面包

??注意：

消费和生产是同时并发并行进行的，不是一次生产一次消费

??实现代码：

public class Bakery { private static int total; //库存public static void main(String[] args) { Producer producer = new Producer(); for(int i = 0; i < 2; i++){ new Thread(producer,"面包师傅-"+(i-1)).start(); } Consumer consumer = new Consumer(); for(int i = 0; i < 10; i++){ new Thread(consumer,"消费者-"+(i-1)).start(); } } private static class Producer implements Runnable{ private int num = 3; //生产者每次生产三个面包 @Override public void run() { try { while(true){ //一直生产 synchronized (Bakery.class){ while((total+num)>100){ //仓库满了，生产者等待 Bakery.class.wait(); } //等待解除 total += num; System.out.println(Thread.currentThread().getName()+"生产面包，库存："+total); Thread.sleep(500); Bakery.class.notifyAll(); //唤醒生产 } Thread.sleep(500); } } catch (InterruptedException e) { e.printStackTrace(); } } } private static class Consumer implements Runnable{ private int num = 1; //消费者每次消费1个面包 @Override public void run() { try { while(true){ //一直消费 synchronized (Bakery.class){ while((total-num)<0){ //仓库空了，消费者等待 Bakery.class.wait(); } //解除消费者等待 total -= num; System.out.println(Thread.currentThread().getName()+"消费面包，库存："+total); Thread.sleep(500); Bakery.class.notifyAll(); //唤醒消费 } Thread.sleep(500); } } catch (InterruptedException e) { e.printStackTrace(); } } } }

部分打印结果：

文章图片

四. 阻塞队列

阻塞队列是一个特殊的队列，也遵循“先进先出”的原则，它是线程安全的队列结构

特性：典型的生产者消费者模型，一般用于做任务的解耦和消峰

队列满的时候，入队列就堵塞等待（生产），直到有其他线程从队列中取走元素
队列空的时候，出队列就堵塞等待（消费），直到有其他线程往队列中插入元素

1. 生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题
生产者和消费者彼此之间不直接通信，而通过阻塞队列来进行通信，所以生产者生产完数据之后等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取
阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力
阻塞队列也能使生产者和消费者之间解耦

上述面包房业务的实现就是生产者消费者模型的一个实例

2. 标准库中的阻塞队列

在 Java 标准库中内置了阻塞队列，如果我们需要在一些程序中使用阻塞队列, 直接使用标准库中的即可

BlockingQueue 是一个接口. 真正实现的类是 LinkedBlockingQueue
put 方法用于阻塞式的入队列, take 用于阻塞式的出队列
BlockingQueue 也有 offer, poll, peek 等方法, 但是这些方法不带有阻塞特性

BlockingDeque queue = new LinkedBlockingDeque<>(); queue.put("hello"); //如果队列为空，直接出出队列就会阻塞 String ret = queue.take(); System.out.println(ret);

3. 阻塞队列的模拟实现这里使用数组实现一个循环队列来模拟阻塞队列

当队列为空的时候，就不能取元素了，就进入wait等待，当有元素存放时，唤醒
当队列为满的时候，就不能存元素了，就进入wait等待，当铀元素取出时，唤醒

??实现代码：

public class MyBlockingQueue { //使用数组实现一个循环队列，队列里面存放的是线程要执行的任务 private Runnable[] tasks; //队列中任务的数量，根据数量来判断是否可以存取 private int count; private int putIndex; //存放任务位置 private int takeIndex; //取出任务位置//有参的构造方法，表示队列容量 public MyBlockingQueue(int size){ tasks = new Runnable[size]; }//存任务 public void put(Runnable task){ try { synchronized (MyBlockingQueue.class){ //如果队列容量满了，则存任务等待 while(count == tasks.length){ MyBlockingQueue.class.wait(); } tasks[putIndex] = task; //将任务放入数组 putIndex = (putIndex+1) % tasks.length; //更新存任务位置 count++; //更新存放数量 MyBlockingQueue.class.notifyAll(); //唤醒存任务 } } catch (InterruptedException e) { e.printStackTrace(); } }//取任务 public Runnable take(){ try { synchronized (MyBlockingQueue.class){ //如果队列任务为空，则取任务等待 while(count==0){ MyBlockingQueue.class.wait(); } //取任务 Runnable task = tasks[takeIndex]; takeIndex = (takeIndex+1) % tasks.length; //更新取任务位置 count--; //更新存放数量 MyBlockingQueue.class.notifyAll(); //唤醒取任务 return task; } } catch (InterruptedException e) { throw new RuntimeException("存放任务出错",e); } } }