多线程-并发编程(7)-生产者消费者模式及非阻塞队列与阻塞队列实现
生产者消费者模式是一个十分经典的多线程协作模式
弄懂生产者消费者问题能够让我们对多线程编程的理解更加深刻
存在3个元素
1.生产者(类比厨师)
2.生产者的生产产品(类比美食)
3.消费者(类比吃货)
思路分析:
理想情况:
最开始,生产者先抢到cpu执行权;生产出生产产品并放在2者位置之间(类比餐桌)
然后消费者抢到cpu执行权,消费掉生产产品
如此反复循环
生产者和消费者简单来说就是2个线程轮流执行
而实际上线程是随机执行的
下面介绍2个异常的情况:
消费者等待过程:
最开始,消费者先抢到cpu执行权;
现在2者之间还未有生产产品,消费者只能在此等待
然后生产者抢到cpu执行权,生产出生产产品,放在2者之间并通知消费者
消费者得到通知,消费掉生产产品
消费者步骤:
1.判断2者之间是否有生产产品
2.如果没有就等待
生产者步骤:
1.生产出生产产品
2.把生产产品放在2者之间
3.通知等待的消费者进行消费
生产者等待过程:
最开始,生产者先抢到cpu执行权;生产出生产产品并放在2者位置之间
然后还是生产者抢到cpu执行权;由于生产产品还未被消费,自然不再进行生产,从而进行等待
然后消费者抢到cpu执行权,消费掉生产产品
生产者步骤:
1.判断2者之间是否有生产产品;如果有就等待,如果没有才生产
2.把生产产品放在2者之间
3.通知等待的消费者进行消费
消费者步骤:
1.判断2者之间是否有生产产品
2.如果没有就等待
3.如果有就消费掉生产产品,2者之间的生产产品就没有了,通知等待的生产者继续生产,生产产品数量减1
代码实现:
此处生产者是厨师,生产产品是汉堡包,消费者是吃货,2者之间是桌子
//主体public class Demo { public static void main(String[] args) { /*消费者步骤: 1,判断桌子上是否有汉堡包。 2,如果没有就等待。 3,如果有就开吃 4,吃完之后,桌子上的汉堡包就没有了 叫醒等待的生产者继续生产 汉堡包的总数量减一*/ /*生产者步骤: 1,判断桌子上是否有汉堡包 如果有就等待,如果没有才生产。 2,把汉堡包放在桌子上。 3,叫醒等待的消费者开吃。*/ Foodie f = new Foodie(); Cooker c = new Cooker(); f.start(); c.start(); }}//生产者(厨师)public class Cooker extends Thread {// 生产者步骤:// 1,判断桌子上是否有汉堡包// 如果有就等待,如果没有才生产。// 2,把汉堡包放在桌子上。// 3,叫醒等待的消费者开吃。 @Override public void run() { while(true){ synchronized (Desk.lock){ if(Desk.count == 0){ break; }else{ if(!Desk.flag){ //生产 System.out.println("厨师正在生产汉堡包"); Desk.flag = true; Desk.lock.notifyAll(); }else{ try {Desk.lock.wait(); } catch (InterruptedException e) {e.printStackTrace(); } } } } } }}//2者之间(桌子)public class Desk { //定义一个标记 //true 就表示桌子上有汉堡包的,此时允许吃货执行 //false 就表示桌子上没有汉堡包的,此时允许厨师执行 public static boolean flag = false; //汉堡包的总数量 public static int count = 10; //锁对象 public static final Object lock = new Object();}//消费者(吃货)public class Foodie extends Thread { @Override public void run() {// 1,判断桌子上是否有汉堡包。// 2,如果没有就等待。// 3,如果有就开吃// 4,吃完之后,桌子上的汉堡包就没有了// 叫醒等待的生产者继续生产// 汉堡包的总数量减一 //套路: //1. while(true)死循环 //2. synchronized 锁,锁对象要唯一 //3. 判断,共享数据是否结束. 结束 //4. 判断,共享数据是否结束. 没有结束 while(true){ synchronized (Desk.lock){ if(Desk.count == 0){ break; }else{ if(Desk.flag){ //有 System.out.println("吃货在吃汉堡包"); Desk.flag = false; Desk.lock.notifyAll(); Desk.count--; }else{ //没有就等待 //使用什么对象当做锁,那么就必须用这个对象去调用等待和唤醒的方法. try {Desk.lock.wait(); } catch (InterruptedException e) {e.printStackTrace(); } } } } } }}
阻塞队列的基本使用:
阻塞队列(BlockingQueue) 是一个支持两个附加操作的队列
这两个附加的操作是:在队列为空时,获取元素的线程会等待队列变为非空。当队列满时,存储元素的线程会等待队列可用
阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素
阻塞队列继承结构:
BlockingQueue的核心方法:
put(anObject):将参数放入队列,如果放不进去会阻塞
take():取出第一个数据,取不到会阻塞
常见的BlockingQueue:
ArrayBlockingQueue:底层是数组,有界
LinkedBlockingQueue:底层是链表,无界;但不是真正的无界,最大为int的最大值
示例代码:
import java.util.concurrent.ArrayBlockingQueue;public class Main { public static void main(String[] a) throws Exception{ // 创建阻塞队列的对象,容量为 1 ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(1); // 存储元素 arrayBlockingQueue.put("汉堡包"); // 取元素 System.out.println(arrayBlockingQueue.take()); System.out.println(arrayBlockingQueue.take()); // 取不到会阻塞 System.out.println("程序结束了");//不会进行这段语句,因为上面语句已经导致阻塞了 }}
阻塞队列实现等待唤醒机制:
代码实现:
//主体public class Demo { public static void main(String[] args) { ArrayBlockingQueue bd = new ArrayBlockingQueue(1); Foodie f = new Foodie(bd); Cooker c = new Cooker(bd); f.start(); c.start();}//生产者(厨师)public class Cooker extends Thread { private ArrayBlockingQueue bd; public Cooker(ArrayBlockingQueue bd) { this.bd = bd; }// 生产者步骤:// 1,判断桌子上是否有汉堡包// 如果有就等待,如果没有才生产。// 2,把汉堡包放在桌子上。// 3,叫醒等待的消费者开吃。 @Override public void run() { while (true) { try { bd.put("汉堡包"); System.out.println("厨师放入一个汉堡包"); } catch (InterruptedException e) { e.printStackTrace(); } } }}//消费者(吃货)public class Foodie extends Thread { private ArrayBlockingQueue bd; public Foodie(ArrayBlockingQueue bd) { this.bd = bd; } @Override public void run() {// 1,判断桌子上是否有汉堡包。// 2,如果没有就等待。// 3,如果有就开吃// 4,吃完之后,桌子上的汉堡包就没有了// 叫醒等待的生产者继续生产// 汉堡包的总数量减一//套路://1. while(true)死循环//2. synchronized 锁,锁对象要唯一//3. 判断,共享数据是否结束. 结束//4. 判断,共享数据是否结束. 没有结束 while (true) { try { String take = bd.take(); System.out.println("吃货将" + take + "拿出来吃了"); } catch (InterruptedException e) { e.printStackTrace(); } } }}