生产者消费者

生产者和消费者模式概述

  • 概述
    生产者消费者模式是一个十分经典的多线程协作的模式,弄懂生产者消费者问题能够让我们对多线程编程的理解更加深刻。
    所谓生产者消费者问题,实际上主要是包含了两类线程:
    一类是生产者线程用于生产数据
    一类是消费者线程用于消费数据
    为了解耦生产者和消费者的关系,通常会采用共享的数据区域,就像是一个仓库
    生产者生产数据之后直接放置在共享数据区中,并不需要关心消费者的行为
    消费者只需要从共享数据区中去获取数据,并不需要关心生产者的行为
  • Object类的等待和唤醒方法
    | 方法名 | 说明 | | —- | —- | | void wait() | 导致当前线程等待,直到另一个线程调用该对象的 notify()方法或 notifyAll()方法 | | void notify() | 唤醒正在等待对象监视器的单个线程 | | void notifyAll() | 唤醒正在等待对象监视器的所有线程 |

生产者和消费者案例

  • 案例需求
    • 桌子类(Desk):定义表示包子数量的变量,定义锁对象变量,定义标记桌子上有无包子的变量
    • 生产者类(Cooker):实现Runnable接口,重写run()方法,设置线程任务
      1.判断是否有包子,决定当前线程是否执行
      2.如果有包子,就进入等待状态,如果没有包子,继续执行,生产包子
      3.生产包子之后,更新桌子上包子状态,唤醒消费者消费包子
    • 消费者类(Foodie):实现Runnable接口,重写run()方法,设置线程任务
      1.判断是否有包子,决定当前线程是否执行
      2.如果没有包子,就进入等待状态,如果有包子,就消费包子
      3.消费包子后,更新桌子上包子状态,唤醒生产者生产包子
    • 测试类(Demo):里面有main方法,main方法中的代码步骤如下
      创建生产者线程和消费者线程对象
      分别开启两个线程
  • 代码实现
    ```java public class Desk {

    //定义一个标记 //true 就表示桌子上有汉堡包的,此时允许吃货执行 //false 就表示桌子上没有汉堡包的,此时允许厨师执行 public static boolean flag = false;

    //汉堡包的总数量 public static int count = 10;

    //锁对象 public static final Object lock = new Object(); }

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(); } } } } } } }

public class Foodie extends Thread { @Override public void run() { // 1,判断桌子上是否有汉堡包。 // 2,如果没有就等待。 // 3,如果有就开吃 // 4,吃完之后,桌子上的汉堡包就没有了 // 叫醒等待的生产者继续生产 // 汉堡包的总数量减一

  1. //套路:
  2. //1. while(true)死循环
  3. //2. synchronized 锁,锁对象要唯一
  4. //3. 判断,共享数据是否结束. 结束
  5. //4. 判断,共享数据是否结束. 没有结束
  6. while(true){
  7. synchronized (Desk.lock){
  8. if(Desk.count == 0){
  9. break;
  10. }else{
  11. if(Desk.flag){
  12. //有
  13. System.out.println("吃货在吃汉堡包");
  14. Desk.flag = false;
  15. Desk.lock.notifyAll();
  16. Desk.count--;
  17. }else{
  18. //没有就等待
  19. //使用什么对象当做锁,那么就必须用这个对象去调用等待和唤醒的方法.
  20. try {
  21. Desk.lock.wait();
  22. } catch (InterruptedException e) {
  23. e.printStackTrace();
  24. }
  25. }
  26. }
  27. }
  28. }
  29. }

}

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();

}

}


<a name="02aa8418"></a>
### 生产者和消费者案例优化

-  需求 
   - 将Desk类中的变量,采用面向对象的方式封装起来
   - 生产者和消费者类中构造方法接收Desk类对象,之后在run方法中进行使用
   - 创建生产者和消费者线程对象,构造方法中传入Desk类对象
   - 开启两个线程
-  代码实现  
```java
public class Desk {

    //定义一个标记
    //true 就表示桌子上有汉堡包的,此时允许吃货执行
    //false 就表示桌子上没有汉堡包的,此时允许厨师执行
    //public static boolean flag = false;
    private boolean flag;

    //汉堡包的总数量
    //public static int count = 10;
    //以后我们在使用这种必须有默认值的变量
   // private int count = 10;
    private int count;

    //锁对象
    //public static final Object lock = new Object();
    private final Object lock = new Object();

    public Desk() {
        this(false,10); // 在空参内部调用带参,对成员变量进行赋值,之后就可以直接使用成员变量了
    }

    public Desk(boolean flag, int count) {
        this.flag = flag;
        this.count = count;
    }

    public boolean isFlag() {
        return flag;
    }

    public void setFlag(boolean flag) {
        this.flag = flag;
    }

    public int getCount() {
        return count;
    }

    public void setCount(int count) {
        this.count = count;
    }

    public Object getLock() {
        return lock;
    }

    @Override
    public String toString() {
        return "Desk{" +
                "flag=" + flag +
                ", count=" + count +
                ", lock=" + lock +
                '}';
    }
}

public class Cooker extends Thread {

    private Desk desk;

    public Cooker(Desk desk) {
        this.desk = desk;
    }
//    生产者步骤:
//            1,判断桌子上是否有汉堡包
//    如果有就等待,如果没有才生产。
//            2,把汉堡包放在桌子上。
//            3,叫醒等待的消费者开吃。

    @Override
    public void run() {
        while(true){
            synchronized (desk.getLock()){
                if(desk.getCount() == 0){
                    break;
                }else{
                    //System.out.println("验证一下是否执行了");
                    if(!desk.isFlag()){
                        //生产
                        System.out.println("厨师正在生产汉堡包");
                        desk.setFlag(true);
                        desk.getLock().notifyAll();
                    }else{
                        try {
                            desk.getLock().wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }
}

public class Foodie extends Thread {
    private Desk desk;

    public Foodie(Desk desk) {
        this.desk = desk;
    }

    @Override
    public void run() {
//        1,判断桌子上是否有汉堡包。
//        2,如果没有就等待。
//        3,如果有就开吃
//        4,吃完之后,桌子上的汉堡包就没有了
//                叫醒等待的生产者继续生产
//        汉堡包的总数量减一

        //套路:
            //1. while(true)死循环
            //2. synchronized 锁,锁对象要唯一
            //3. 判断,共享数据是否结束. 结束
            //4. 判断,共享数据是否结束. 没有结束
        while(true){
            synchronized (desk.getLock()){
                if(desk.getCount() == 0){
                    break;
                }else{
                    //System.out.println("验证一下是否执行了");
                    if(desk.isFlag()){
                        //有
                        System.out.println("吃货在吃汉堡包");
                        desk.setFlag(false);
                        desk.getLock().notifyAll();
                        desk.setCount(desk.getCount() - 1);
                    }else{
                        //没有就等待
                        //使用什么对象当做锁,那么就必须用这个对象去调用等待和唤醒的方法.
                        try {
                            desk.getLock().wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }

    }
}

public class Demo {
    public static void main(String[] args) {
        /*消费者步骤:
        1,判断桌子上是否有汉堡包。
        2,如果没有就等待。
        3,如果有就开吃
        4,吃完之后,桌子上的汉堡包就没有了
                叫醒等待的生产者继续生产
        汉堡包的总数量减一*/

        /*生产者步骤:
        1,判断桌子上是否有汉堡包
        如果有就等待,如果没有才生产。
        2,把汉堡包放在桌子上。
        3,叫醒等待的消费者开吃。*/

        Desk desk = new Desk();

        Foodie f = new Foodie(desk);
        Cooker c = new Cooker(desk);

        f.start();
        c.start();

    }
}

3.4阻塞队列基本使用

  • 阻塞队列继承结构
    06_阻塞队列继承结构.png
  • 常见BlockingQueue:
    ArrayBlockingQueue: 底层是数组,有界
    LinkedBlockingQueue: 底层是链表,无界.但不是真正的无界,最大为int的最大值
  • BlockingQueue的核心方法:
    put(anObject): 将参数放入队列,如果放不进去会阻塞
    take(): 取出第一个数据,取不到会阻塞
  • 代码示例

    public class Demo02 {
     public static void main(String[] args) throws Exception {
         // 创建阻塞队列的对象,容量为 1
         ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue<>(1);
    
         // 存储元素
         arrayBlockingQueue.put("汉堡包");
    
         // 取元素
         System.out.println(arrayBlockingQueue.take());
         System.out.println(arrayBlockingQueue.take()); // 取不到会阻塞
    
         System.out.println("程序结束了");
     }
    }
    

3.5阻塞队列实现等待唤醒机制

  • 案例需求
    • 生产者类(Cooker):实现Runnable接口,重写run()方法,设置线程任务
      1.构造方法中接收一个阻塞队列对象
      2.在run方法中循环向阻塞队列中添加包子
      3.打印添加结果
    • 消费者类(Foodie):实现Runnable接口,重写run()方法,设置线程任务
      1.构造方法中接收一个阻塞队列对象
      2.在run方法中循环获取阻塞队列中的包子
      3.打印获取结果
    • 测试类(Demo):里面有main方法,main方法中的代码步骤如下
      创建阻塞队列对象
      创建生产者线程和消费者线程对象,构造方法中传入阻塞队列对象
      分别开启两个线程
  • 代码实现
    ```java 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<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();
        }
    }

}

}

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();
}

} ```