生产者-消费者模型详解

🍎前言🍎

生产者-消费者模式是多线程编程中常见的一种模式，它用于解决生产者和消费者之间的协作问题。生产者负责生成数据，消费者负责处理数据，通过合理的协作，可以实现高效的数据处理。本文将详细介绍 Java 生产者-消费者模式，包括其基本概念、常见用法以及注意事项。

目录

一.为什么要有生产者消费者模型

二.阻塞队列

（一）阻塞队列是什么

（二）模拟实现阻塞队列

📣第一步：先实现一个简单的循环队列（不考虑阻塞情况）

?📣第二步：实现阻塞的功能

三.实现生产者消费者模型?

---

一.为什么要有生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。
也就是通过一个容器来解决生产者和消费者之间彼此关系太过紧密的问题。

💡举例来说：

? ? ? ? 假如有以下的客户端-服务器模型：

如果在极端情况下，客户端发出的请求非常多，此时服务器A响应了很多，就会立即发送给服务器B和服务器C。A这边抗多少访问量，此时B和C 就 完全一样。但是我们要明确，不同的服务器，承担的并发量不一样。有可能A承担这些并发量没有关系，但是 B 和 C承担的话就会服务器崩溃。

二.阻塞队列

在 Java 标准库中内置了阻塞队列. 如果我们需要在一些程序中使用阻塞队列, 直接使用标准库中的即可. 但是此博客会带着大家去模拟实现一个.
BlockingQueue 是一个接口. 真正实现的类是 LinkedBlockingQueue.
put 方法用于阻塞式的入队列, take 用于阻塞式的出队列.
BlockingQueue 也有 offer, poll, peek 等方法, 但是这些方法不带有阻塞特性.

（一）阻塞队列是什么

阻塞队列是一种特殊的队列. 也遵守 "先进先出" 的原则.

阻塞队列的一个典型应用场景就是 "生产者消费者模型". 这是一种非常典型的开发模型.
阻塞队列能是一种线程安全的数据结构, 并且具有以下特性:
????????当队列满的时候, 继续入队列就会阻塞, 直到有其他线程从队列中取走元素.
????????当队列空的时候, 继续出队列也会阻塞, 直到有其他线程往队列中插入元素.

我们这里使用阻塞队列来解决生产者消费者之间的问题：

此时假如客户端发出了超级多的请求，服务器A能够接受，然后A把这些数据写入到阻塞队列中。B可以按照自己的节奏来从阻塞队列中取得请求并且处理。

假设A最多能承受每一秒3000次请求，B每一秒最多能处理1000次请求。

正常情况下， A 和 B 每一秒处理1000次请求。

极端情况下，A这边要处理3000次数据。如果让B也处理3000次请求，那么B就会崩溃。

此时队列就可以帮助B来承担一些数据，比如先承担2000次数据，然后B的从队列中取得自己能承受的请求，慢慢的处理。 虽然这可能会比较慢，但是总好比把 B 给搞崩溃了好！！！

不过，想上述这样的峰值极端情况，一般不会持续存在，只会短时间内出现。过了峰值之后，A的请求量就恢复正常了，B就可以把积压的数据处理掉了。

总结：阻塞队列有两个作用

? ? ? ? 1.解耦合? ?使得 两个 主体之间的关系不是那么密切，发生问题后不会牵涉很多。

? ? ? ? 2.削峰填谷? ?使得大批量的数据请求放到一个阻塞队列中，其他服务器慢慢接收

（二）模拟实现阻塞队列

俗话说的好，自己动手，丰衣足食。如果我们可以把阻塞队列自己实现出来，那么我们对阻塞队列和生产者消费者模型都可以有更好的理解和掌握。

📣第一步：先实现一个简单的循环队列（不考虑阻塞情况）

为什么要设置为循环队列 是因为这个阻塞队列是要不断重复使用的。总不能说用一次扔了！

设计代码如下：

public class MyBlockingQueue {
    private String[] data = new String[1000];
    //队头元素
    private  int head = 0;
    //队尾元素
    private int tail = 0;
    private int size = 0;

    //核心方法,入队和出队
    public void put(String elem) throws InterruptedException {
        synchronized (this) {
            if(size == data.length) {
                //如果队列这时候满了,那就阻塞
                //不出意外,只有take才可以唤醒它

            }
            //队列没有满,此时要往里面添加元素
            data[tail] = elem;
            tail++;
            //如果 tail自增之后 来到了数组的末尾,这个时候需要让他回到开头
            if (tail == data.length) {
                tail = 0;
            }
            size++;
        }
    }

    public String take() throws InterruptedException {
        synchronized (this) {
            //如果此时队列为空
            if(size == 0) {
                //如果为空,那就不能在出元素了,阻塞直到 put 操作放入一个元素

            }
            //如果此时不为空,就要往外出元素
            String ret = data[head];
            head++;
            if (head == data.length) {
                head = 0;
            }
            size--;
            return ret;
        }
    }
}

由于考虑的是并发编程，所以要保证 线程安全。此时就需要加入 synchronized 锁。确保一个线程使用 take 的时候，其他线程不会中途过来捣乱。

此时已经实现了一个简单的循环队列，只不过还没有对阻塞的操作进行实现。4

?📣第二步：实现阻塞的功能

考虑到阻塞，那就要设计到 wait 和 notify 。wait可以阻塞一个线程，而notify可以唤醒一个线程

我们总体的设计愿景是这样的：

1. ?一个队列, 要么是空，要么是满 。
2. ?take 和 put 只能有一边阻塞.
3. 如果put 阻塞了（也就是满了）,其他线程再进行put 也都会被阻塞，只有take可以唤醒（出列）
4. 如果take阻塞了（也就是空了),其他线程再进行take 也都会被阻塞，只有put可以唤醒（入队）

那么我们可以设计出如下的代码：

public class MyBlockingQueue {
    private String[] data = new String[1000];
    //队头元素
    private  int head = 0;
    //队尾元素
    private int tail = 0;
    private int size = 0;

    //核心方法,入队和出队
    public void put(String elem) throws InterruptedException {
        synchronized (this) {
            while (size == data.length) {
                //如果队列这时候满了,那就阻塞
                //不出意外,只有take才可以唤醒它
                this.wait();
            }
            //队列没有满,此时要往里面添加元素
            data[tail] = elem;
            tail++;
            //如果 tail自增之后 来到了数组的末尾,这个时候需要让他回到开头
            if (tail == data.length) {
                tail = 0;
            }
            size++;
            this.notify();// 此时加入了一个元素,那么take中的wait就需要被唤醒
        }
    }

    public String take() throws InterruptedException {
        synchronized (this) {
            //如果此时队列为空
            while (size == 0) {
                //如果为空,那就不能在出元素了,直到 put 操作放入一个元素
                this.wait();
            }
            //如果此时不为空,就要往外出元素
            String ret = data[head];
            head++;
            if (head == data.length) {
                head = 0;
            }
            size--;
            this.notify();// 此时出去了一个元素, put 中的 wait 需要被唤醒
            return ret;
        }
    }
}

代码解读：

1. 在多线程环境下。举例来看put方法，假如此时的队列已经满了，然后this.wait()阻塞，此时只能通过 take 取出元素的操作来给队列腾出一个空间，同时要唤醒 put 中的wait，让其他线程可以调用 put ，往里面放入元素。

2. 为什么要使用while 循环来进行判断 而不是 if 呢？此时 一定要考虑是 notify 唤醒的 还是 interrupt 唤醒的，如果是通过 interrupt 唤醒的，说明此处可能不是通过put 或者 take 方法进行调用的，此时的队列还是满的，再次进行代码操作，就有可能覆盖掉原来的有效数据。

所以使用while 循环，可以在 wait 唤醒之后再次进行判断。以防万一！

三.实现生产者消费者模型?

代码入下：

public class Demo1 {
    public static void main(String[] args) {
        /**
         * 此测试假如是 生产的快,消费的慢                 
         */
        MyBlockingQueue queue = new MyBlockingQueue();
        Thread t1 = new Thread(()-> {
            //消费者 消费很多数字
            while (true) {
                try {
                    String result = queue.take();
                    System.out.println("消费元素: " + result);
                    Thread.sleep(500);  //是为了产生速度差，消费的满
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

        });

        Thread t2 = new Thread(()-> {
            //生产者负责生产很多数字
            int num = 1;
            while (true) {
                try {
                    queue.put(num + " ");
                    System.out.println("生产元素: " + num);
                    num++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t1.start();
        t2.start();
    }
}

运行结果如下：

可以看到，由于此时消费者正在 sleep，生产者这时候生产了很多数字，以至于阻塞队列都满了。消费者一次只能从阻塞队列中消费一个数据，那么我们可以看到，此时生产者的速度就满了下来，完全是按照消费者的节奏来走的。 此时就简单的模拟实现了一个生产者-消费者模型。

---

总结：生产者消费者模型在生活中也经常使用，在工作中更是重点！如有不足，请指正！
?