【多线程】深入剖析生产者-消费者模型

💐个人主页：初晴~

📚相关专栏：多线程 / javaEE初阶

一、阻塞队列

阻塞队列是⼀种特殊的队列，也遵守 "先进先出" 的原则。是在普通的队列基础上做出了补充。

java标准库中的原有的队列Queue及其子类，默认都是线程不安全的，而阻塞队列能是⼀种线程安全的数据结构，具有阻塞特性：

当队列满的时候, 继续⼊队列就会阻塞, 直到有其他线程从队列中取⾛元素
当队列空的时候, 继续出队列也会阻塞, 直到有其他线程往队列中插⼊元素

基于阻塞队列最大的应用场景就是“生产消费者模型”了。

二、生产者消费者模型

1、概念

比如一个生产线需要制作零件，组装两个步骤，这时就由两种方式进行生产部署：

1、每个人都分别进行零件制作和组装两个操作

2、将员工分成两拨人，一拨人专门负责制作零件，另一拨人专门负责组装

显然第一种方法是比较低效的，制作零件的机器数量是有限的，所有人就会去竞争这几台机器，可能会导致阻塞等待导致效率低下。而第二种方法则实现了各司其职，自然效率会更高，这时零件制作就相当于是生产者，组装就相当于是消费者，这种方式就被称之为“生产者消费者模型”，在后端开发中经常会涉及到。

2、作用

⽣产者消费者模式就是通过⼀个容器来解决⽣产者和消费者的强耦合问题。

⽣产者和消费者彼此之间不直接通讯，⽽通过阻塞队列来进⾏通讯，所以⽣产者⽣产完数据之后不⽤等待消费者处理，直接扔给阻塞队列，消费者不找⽣产者要数据，⽽是直接从阻塞队列⾥取.

1. 阻塞队列能使⽣产者和消费者之间解耦合.

直接耦合的代码结构：

这样A与BC的耦合度非常高，出一点问题就可能导致程序崩溃，并且对于后期维护来说也十分不方便。

引入生产者消费者模型的代码结构：

这样显然A与B直接就解耦合了，不会直接关联。

注意：

阻塞队列是一种 数据结构，由于比较好用，会将其单独封装为一个服务器程序，并且在单独的服务器机器上进行部署。这时的阻塞队列就被称之为 “消息队列”（Messae Queue，MQ）了。而消息队列相对是比较成熟的，代码不会频繁修改，因此我们认为A与队列，B与队列之间的交互是 低耦合的。

2. 阻塞队列就相当于⼀个缓冲区，平衡了⽣产者和消费者的处理能⼒. (削峰填⾕)

⽐如在 "秒杀" 场景下, 服务器同⼀时刻可能会收到⼤量的⽀付请求. 如果直接处理这些⽀付请求, 服务器可能扛不住(每个⽀付请求的处理都需要⽐较复杂的流程). 这个时候就可以把这些请求都放到⼀个阻塞队列中, 然后再由消费者线程慢慢的来处理每个⽀付请求.

这样做可以有效进⾏ "削峰", 防⽌服务器被突然到来的⼀波请求直接冲垮.

（1）为什么一个服务器请求收到过多就有可能会崩溃？

一台服务器就是一台“电脑”，提供了一些硬件资源（CPU、内存、硬盘、网络带宽……），服务器每次收到一个请求，在处理的过程中就需要执行一系列代码，执行代码的过程中就需要 消耗一定的硬件资源，而服务器的硬件资源是有限的，当请求需要消耗的硬件资源的量 超过机器上限时，机器就会出现问题（卡死、程序崩溃……）

（2）当请求激增时，为什么A与消息队列不会崩溃？

A相当于是一个 “网关服务器”，负责接收客户端的请求，再把请求发给其它服务器。这样的服务器做的工作是比较简单的（ 单纯的数据转发），处理同样一个请求消耗的硬件资源更少，在同等情况下能处理更多请求。消息队列中的程序也比较简单，消耗的资源也相对较少，因此他们都不太容易会崩溃。

而B是真正处理业务逻辑的服务器，代码量更加庞大， 消耗的时间与硬件资源更多，当请求激增时就更容易崩溃了。

优点：

1、解耦合

2、削峰填谷

缺点：

1、需要更多的机器来部署消息队列

2、生产者与消费者之间的通信会延时，响应时间会变长

三、实现方式

1、手写一个阻塞队列

在之前的文章中，我们曾在栈和队列一文深入探讨过队列的写法，这里就不做过多赘述，直接看一下最基础的队列的代码实现：

java">class MyBlockingQueue{
    private String[] data=null;
    private int head=0;
    private int tail=0;
    private int size=0;
    public MyBlockingQueue(int capacity){
        data=new String[capacity];
    }

    public void put(String s){
        if(size==data.length){
            //队列满了
            return;
        }
        data[tail]=s;
        tail++;
        if(tail>=data.length){
            tail=0;
        }
        size++;
    }

    public String take(){
        if(size==0){
            //队列为空
            return null;
        }
        String ret=data[head];
        head++;
        if(head>=data.length){
            head=0;
        }
        size--;
        return ret;
    }
}

由于put与take方法中涉及了很多的修改操作，这样的代码在多线程环境下肯定是会有线程安全问题的，博主在深入剖析线程安全问题一文中做过详细分析。

那么该如何解决这一问题呢，显然就是通过加锁操作了：

但这样还是不够的。当线程发现队列为空时，不应该继续再去参与锁的竞争应该直接进入阻塞等待的状态，等到其它线程调用put方法，队列不为空时，再恢复执行，从而避免浪费不必要的资源，提高执行效率，这时就可以利用wait-notify来解决了。这在博主的等待通知机制一文过详细的介绍。于是最终代码实现就如下：

java">class MyBlockingQueue{
    private String[] data=null;
    private int head=0;
    private int tail=0;
    private int size=0;
    public MyBlockingQueue(int capacity){
        data=new String[capacity];
    }

    public void put(String s){
        synchronized (this){
            if(size==data.length){
                //队列满了
                return;
            }
            data[tail]=s;
            tail++;
            if(tail>=data.length){
                tail=0;
            }
            size++;
            this.notify();
        }
    }

    public String take() throws InterruptedException {
        synchronized (this){
            if(size==0){
                //队列为空
                this.wait();
            }
            String ret=data[head];
            head++;
            if(head>=data.length){
                head=0;
            }
            size--;
            return ret;
        }
    }
}

2、实现生产者消费者模型

在 Java 标准库中内置了阻塞队列。如果我们需要在⼀些程序中使⽤阻塞队列, 直接使⽤标准库中的BlockingQueue即可。

BlockingQueue 是⼀个接⼝. 真正实现的类LinkedBlockingQueue/ArrayBlockingQueue等
put ⽅法⽤于阻塞式的⼊队列, take ⽤于阻塞式的出队列
BlockingQueue 也有 offer, poll, peek 等⽅法, 但是这些⽅法不带有阻塞特性

使用示例：

java">public class Main {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue=new ArrayBlockingQueue<>(3);
        queue.put("111");
        System.out.println("put 成功");
        queue.put("111");
        System.out.println("put 成功");
        
        queue.take();
        System.out.println("take 成功");
        queue.take();
        System.out.println("take 成功");
        queue.take();
        System.out.println("take 成功");
    }
}

我们可以看到当执行第三个take时，由于此时队列为空，因此线程就会进入阻塞状态了。

接着我们就可以用它来试着简单实现一个生产者消费者模型了：

java">public class Main{
    public static void main(String[] args) {
        BlockingQueue<Integer> queue=new ArrayBlockingQueue<>(1000);
        //生产者线程
        Thread t1=new Thread(()->{
            int i=1;
            while(true){
                try {
                    queue.put(i);
                    System.out.println("生产元素 "+ i);
                    i++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        
        //消费者线程
        Thread t2=new Thread(()->{
            while (true){
                try {
                    Integer i=queue.take();
                    System.out.println("消费元素 "+ i);
                    
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        
        t1.start();
        t2.start();
    }
}

其中t1就作为生产者线程，t2作为消费者线程，让我们试试看让生产慢于消费会发生什么：