多线程-并发编程-生产者消费者模式及非阻塞队列与阻塞队列实现

Posted 2022-04-18 汤键.TJ

生产者消费者模式是一个十分经典的多线程协作模式
弄懂生产者消费者问题能够让我们对多线程编程的理解更加深刻

存在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<String> 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<String> 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<String> bd;
    public Cooker(ArrayBlockingQueue<String> 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<String> bd;
    public Foodie(ArrayBlockingQueue<String> 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();