多线程生产者消费者模型

我不是女神ヾ 2023-02-14 13:10 13阅读 0赞

# 1. 基础知识： #

## 1. 什么是生产者-消费者模式： ##

比如有两个进程A和B，它们共享一个固定大小的缓冲区，A进程产生数据放入缓冲区，B进程从缓冲区中取出数据进行计算，那么这里其实就是一个生产者和消费者的模式，A相当于生产者，B相当于消费者。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70]

## 2. 为什么要使用生产者消费者模式： ##

在多线程开发中，如果生产者生产数据的速度很快，而消费者消费数据的速度很慢，那么生产者就必须等待消费者消费完了数据才能够继续生产数据，因为生产那么多也没有地方放啊；同理如果消费者的速度大于生产者那么消费者就会经常处理等待状态，所以为了达到生产者和消费者生产数据和消费数据之间的平衡，那么就需要一个缓冲区用来存储生产者生产的数据，所以就引入了生产者-消费者模式。

简单来说这里的缓冲区的作用就是为了平衡生产者和消费者的处理能力，起到一个数据缓存的作用，同时也达到了一个解耦的作用。

## 3. 生产者-消费者模式的特点： ##

1.  保证生产者不会在缓冲区满的时候继续向缓冲区放入数据，而消费者也不会在缓冲区空的时候，消耗数据；
2.  当缓冲区满的时候，生产者会进入休眠状态，当下次消费者开始消耗缓冲区的数据时，生产者才会被唤醒，开始往缓冲区中添加数据；当缓冲区空的时候，消费者也会进入休眠状态，直到生产者往缓冲区中添加数据时才会被唤醒；  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 1]

# 2. 问题要点： #

其实生产者消费者模型挺像观察者模式的，对于该模型我们应该明确以下4点：

1. 当生产者生产出产品时，应该通知消费者去消费。
    2. 当消费者消费完产品，应该通知生产者去生产。
    3. 当产品的库存满了的时候，生产者不应该再去生产，而是通知消费者去消费。
    4. 当产品的库存为0的时候，消费者不应该去消费，而是通知生产者去生产。

生产者和消费者问题是线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。  
![在这里插入图片描述][20200602234010942.png]  
![在这里插入图片描述][2020060223411627.png]

# 3. 代码实现：wait()和notify()实现 #

**利用内部线程之间的通信：Object的wait() / notify()方法来实现生产者-消费者模型。  
ps：采用wait()/notify()方法的缺点是不能实现单生产者单消费者模式，因为要是用notify()就必须使用同步代码块。**

## 1. 定义生产者。 ##

Producter.java

package demo;
    
    public class Producter implements Runnable { 
    
        private Storage resource;
    
        public Producter(Storage resource) { 
            super();
            this.resource = resource;
        }
    
        @Override
        public void run() { 
            while (true) { 
                try { 
                    Thread.sleep(100);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
    
                resource.produce();
            }
        }
    }

## 2. 定义消费者。 ##

Consumer.java

package demo;
    
    public class Consumer implements Runnable { 
    
        private Storage resource;
    
        public Consumer(Storage resource) { 
            super();
            this.resource = resource;
        }
    
        @Override
        public void run() { 
            while (true) { 
                try { 
                    Thread.sleep(500);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                resource.cosume();
            }
        }
    }

## 3. 定义Storage仓库。 ##

Storage.java

package demo;
    
    public class Storage { 
    
        private int MAX_SIZE = 20;
    
        private int count = 0;
    
        public synchronized void cosume() { 
            while (count == 0) { 
                try { 
                    System.out.println("【消费者】库存已经为空了，暂时不能进行消费任务!");
                    wait();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
            count--;
            System.out.println("【消费者】" + Thread.currentThread().getName() + "消费产品, 库存:" + count);
            this.notify();
        }
    
        public synchronized void produce() { 
            while (count >= MAX_SIZE) { 
                try { 
                    System.out.println("【生产者】库存已经满了，暂时不能进行生产任务！");
                    wait();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
            count++;
            System.out.println("【生产者】" + Thread.currentThread().getName() + "生产产品, 库存" + count);
            this.notify();
        }
    }

## 4. 测试demo。 ##

ProducterCosumerXDemo.java

package demo;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class ProducterCosumerXDemo { 
    
        public static void main(String[] args) { 
    
            Storage storage = new Storage();
            ExecutorService service = Executors.newCachedThreadPool();
    
            for (int i = 0; i < 5; i++) { 
                service.submit(new Producter(storage));
            }
    
            for (int i = 0; i < 5; i++) { 
                service.submit(new Consumer(storage));
            }
    
            service.shutdown();
        }
    }

# 4. 运行结果： #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 4]

# 5. 分析： #

**我们这里创建了5个生产者，5个消费者。生产者生产的速度比消费者消费的速度要快，从图中很明显看到生产者率先生产出20个产品，已是库存极大值，往后不能再去生产了，然后通知消费者去消费。**

# 6. 参考： #

1.  [关于Java多线程的一些常考知识点][Java]
2.  [Java 实例 - 生产者/消费者问题][Java _ -]
3.  [经典并发同步模式：生产者-消费者设计模式][-]
4.  [Java实现生产者-消费者模型][Java_-]
5.  [Java实现生产者-消费者模型][Java_- 1]
6.  [Java 生产者消费者实现 —— BlockingQueue][Java _ _ BlockingQueue]
7.  [Java实现生产者和消费者的5种方式][Java_5]
8.  [IntelliJ IDEA 如何创建一个普通的 Java 项目，及创建 Java 文件并运行][IntelliJ IDEA _ Java _ Java]
9.  [Java 接口][Java 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020060223353041.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200602233833843.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ==,size_16,color_FFFFFF,t_70
[20200602234010942.png]: https://img-blog.csdnimg.cn/20200602234010942.png
[2020060223411627.png]: https://img-blog.csdnimg.cn/2020060223411627.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200602232719483.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200602232728496.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200602232740538.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5eF9seQ==,size_16,color_FFFFFF,t_70
[Java]: https://cloud.tencent.com/developer/article/1143577
[Java _ -]: https://www.runoob.com/html/thread-procon.html
[-]: https://zhuanlan.zhihu.com/p/73442055
[Java_-]: https://juejin.im/post/5a123bb251882503dc534e59
[Java_- 1]: https://monkeysayhi.github.io/2017/10/08/Java%E5%AE%9E%E7%8E%B0%E7%94%9F%E4%BA%A7%E8%80%85-%E6%B6%88%E8%B4%B9%E8%80%85%E6%A8%A1%E5%9E%8B/
[Java _ _ BlockingQueue]: https://www.jianshu.com/p/a42b89287359
[Java_5]: https://juejin.im/entry/596343686fb9a06bbd6f888c
[IntelliJ IDEA _ Java _ Java]: https://blog.csdn.net/oschina_41790905/article/details/79475187
[Java 1]: https://www.runoob.com/java/java-interfaces.html