生产者消费者模型（能看懂文字就能明白系列）

本文主要是介绍生产者消费者模型（能看懂文字就能明白系列），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

系列文章目录

能看懂文字就能明白系列
C语言笔记传送门
Java笔记传送门
🌟 个人主页：古德猫宁-

🌈 信念如阳光，照亮前行的每一步

前言

本节目标：

理解什么是阻塞队列，阻塞队列与普通队列的区别
理解什么是生产者消费者模型
生产者消费者模型的主要作用

一、阻塞队列

阻塞独立是一个特殊的队列，它具有以下特点：

线程安全
带有阻塞特性：即如果队列为空，这时继续出队列的话，会发生阻塞，阻塞到其他线程往队列添加元素为止。如果队列为满，这时继续入队列的话，也会发生阻塞，阻塞到其他线程从队列取出元素为止。

利用上述特性，我们可以使用阻塞队列来实现一个“生产者消费者模型“

二、什么是生产者消费者模型

举个生活中的例子，比如有A B C三个人，每个人分别负责做饺子皮，包饺子，但是做饺子皮的工具只有一个，A在做饺子皮的时候，别人只能看着A干活，这时候效率是不高的，我们可以让A专门只负责做饺子皮这一工作，B和C两人负责包饺子，这种干活模式就不会设计到“工具”的竞争，这时候A就会不停的生产饺子皮，B和C会不停的消耗饺子皮。

而A生产出的饺子皮必须要有一个地方存放着，这个“地方”就叫做阻塞队列。

A就是生产者，把生产出来的内容放到阻塞队列中。

B和C就是消费者，会从阻塞队列中获取内容并消费。

如果A生产的比较慢，B和C就得等待（因为他们从“空的队列”中获取元素就会阻塞）

如果A生产的比较快，A就得等待（因为往“满的队列”中添加元素也会阻塞）

三、生产者消费者模型的意义

解耦合：两个模块，联系越紧密，耦合就越高，尤其是对于分布式系统来说，是更加有意义的。

如上图所示，如果服务器A和服务器B直接交互（A把请求发送给B，B把响应返回A），这样彼此之间的耦合是比较高的。
如果B出现问题，A很可能也会被影响到。
如果未来再添加一个服务器C，就需要对服务器A这边的代码做出一定的改动。

相比之下，如果采用生产者消费者模型，可以有效解决这种耦合问题
在这里插入图片描述
此时，耦合就会被降低，如果服务器B这边出现问题，就不会对服务器A产生直接影响.（服务器A只是和队列交互，不知道服务器B的存在）
后续如果新增一个服务器C，此时，服务器A不用进行任何修改，只需要让服务器C从队列中获取数据即可.

削峰填谷
这里的峰和谷：
峰：短时间内请求量比较大
谷：请求量比较小

在这个结构下，一旦客户端这边发起的请求非常多了，服务器A收到的每一个请求，都会立即发给服务器B，服务器A这边抗多少访问量，服务器B和服务器A完全一样。
不同的服务器，上面跑到业务不同，虽然访问量一样，单个访问，消耗的硬件资源不一样，可能服务器A承担这些并发量没啥事，服务器B承担这些并发量就会挂了。（比如数据库本身就是一个分布式系统，相对脆弱）

引入生产者消费者模型，上述问题也可以被改善。
在这里插入图片描述
服务器A这边收到了较大的请求量，服务器A会把对应的请求写入到队列中，服务器B仍然可以按照之前的节奏，来处理请求。

比如，正常情况下，服务器A和服务器B每秒钟处理1000次请求，极端情况下，服务器A这边每秒要处理3000次请求，如果让服务器B也处理3000次，服务器B就要挂了，阻塞队列帮服务器B承担了压力，服务器B仍然可以按照1000次的节奏处理请求。

像上述的峰值情况，一般不会持续存在，只会短时间出现，过了峰值之后，服务器A的请求量就恢复正常了，服务器B就可以逐渐把挤压的数据都给处理掉了。

四、代码实现

public class blockingQueue {private String[] data = new String[1000];private volatile int head = 0;private volatile int rear = 0;private volatile int size = 0;public void put(String elem) throws InterruptedException {synchronized (this) {while (size == data.length) {//队列满了//如果是普通的队列，满了就直接return即可this.wait();}//队列没满，往里面添加元素data[rear] = elem;rear++;if (rear == data.length) {rear = 0;//如果rear自增之后，到达了队列结尾，这个时候需要让他回到开头（环形队列）}size++;//这个notify用来唤醒take中的waitthis.notify();}}public String take() throws InterruptedException {synchronized (this) {while (size == 0) {//队列空了//对于普通队列，直接返回即可this.wait();}//队列不空，就可以把对首元素（head位置的元素）删除掉，并进行返回String ret = data[head];head++;if (head == data.length) {head = 0;}size--;//这个notify用来唤醒put中的wait，表示当前队列不是满的，可以继续添加元素了this.notify();return ret;}}public static void main(String[] args) {blockingQueue queue = new blockingQueue();//消费者Thread t1 = new Thread(() -> {while (true) {try {String result = queue.take();System.out.println("消费元素:" + result);} catch (InterruptedException e) {e.printStackTrace();}}});//生产者Thread t2 = new Thread(()->{int num = 1;while (true){try {queue.put(num+" ");} catch (InterruptedException e) {e.printStackTrace();}num++;System.out.println("生产元素："+num);try {Thread.sleep(500);} catch (InterruptedException e) {e.printStackTrace();}}});t1.start();t2.start();}
}