🍉 阻塞队列

在学习数据结构时 我们了解到了什么是队列,普通队列遵守先进先出的规则,阻塞队列同样也符合这样的规则,但是相比于普通队列阻塞队列还有着其他的功能

🍉阻塞队列的功能

• 线程安全:阻塞队列是我们多线程中的案列之一,既然属于多线程,那么一定和线程安全有关系,阻塞队列主要的功能就是能保证线程安全
• 产生阻塞效果
  1.若队列为空,尝试出队列,就会出现阻塞,阻塞到队列不为空为止
  2 .若队列为满,尝试如队列也会出现堵塞,阻塞到队列不为满为止.
  基于上述问题,我们就可以实现一个典型的生产者和消费者模型

🍉生产者消费者模型

我们举个场景🌰 :
过年的时候,大家家里都会包饺子,一家三口都会齐上阵,为了高效率的包,一般都会分工,假如👨擀饺子皮,👦和👩包饺子,这样效率会大大提高,👨擀完饺子皮之后会把饺子皮放在盖帘上,然后👦和👩将盖帘子上的饺子皮包成饺子,这就是一个典型的生产者消费者模型
生产者:👨
消费者:👦和👩
这里的盖帘子我们把他叫做交易场所

盖帘子上饺子皮太多,会导致生产者阻塞,盖帘子没有更多的饺子皮,会导致消费者堵塞

生产者和消费者模型是实际开发中非常有用的一种多线程开发手段,尤其是在服务器开发中
假设有两个服务器A,B A作为入口服务器作为直接接受用户的网络请求,B作为应用服务器,来给A提供一些数据

🍉生产者消费者模型的优点

1. 优点一:能够让多个服务器之间充分的解耦合
   
   何为解耦合?
   打个比方:如果不使用生产者消费者模型,就像上面的图所示,这样的话A,B两个服务器耦合性是非常高的,在开发A是,需要充分的了解到B的接口是什么样的,开发B的时候也需要知道A是怎么调用的,一旦把B换成C,A是需要进行非常大的改动的,而且如果B挂了A也顺带着挂了,显然在计算机中这样的代码是非常不可取的,耦合性太高,在日常开发中我也需要尽量做到高内聚,低耦合

如果我们引进生产者消费者模型,就解决了这个问题

如图

当我们引进了阻塞队列后,就会产生两个生产者消费者模型
1. 对于请求:A是生产者,B是消费者
2.对于响应:A是消费者,B是生产者
阻塞队列就是A和B的交易场所

A和B分别和阻塞队列进行交互,A不需要认识B,B也不需要认识A,队列是不会变的 B挂了,A也没啥影响,把B换成C,A也完全感知不到,这就达到了我们所说的低耦合的理念

2. 优点二:能够对请求进行削峰填谷

如图,A作为入口服务器,计算量很轻,即使请求暴涨,A服务器也不会奔溃,但是B作为响应服务器,计算量很大,所需要的资源也很多,如果请求量暴涨,B就会挂掉
但是如果引进阻塞队列的话,就会有所改变

引进阻塞队列后就会起到削峰填谷的效果

1. 削峰

加入阻塞队列后,A将所有的请求放进了阻塞队列里,阻塞队列没有什么计算量,只是存储数据,所以几乎能容纳所有的A传过来的请求,B这边不会一下子增长很多请求,他会按照原来的速度处理数据,B就会得到很好的保护,不会因为请求的数据暴涨而崩溃,这就是削峰

2. 填谷

像这种数据突然增长许多的情况只会维持很多的一段时间,等峰值过去后,数据增长又恢复了,B服务器不会因为请求的减少而闲下来,仍然会按照原有的速度通过阻塞队列来处理数据,这就叫做填谷

以上就是削峰填谷,其实这样的功能主要是为了保护像B这样的响应/应用服务器,保证程序不会因为数据的暴涨而崩溃

在现实生活中也有这样的🌰 :三峡大坝,不懂得小伙伴可以去了解一下

🥝阻塞队列的实现

🥝 Java中标准库的阻塞队列

我们先根据Java中内置的阻塞队列,基于他来实现一个生产者消费者模型,再来实现自己创建的阻塞队列

 public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>();
        blockingQueue.put("hello");
        blockingQueue.take();
    }

这就是一个简单的阻塞队列

🥝自己实现一个阻塞队列

1. 先实现一个简单的队列

队列可以由数组和链表组成,数组实现起来呢更加简单,所以我们用数组实现,并且使用循环数组
1.设置三个变量分别的链表的头(head),尾(end),还有链表的长度(size);
2.入队列,把元素放在end的位置上,并且end++,size++;,如果end的值等于数组的长度,end设置为0;
3.出队列,head++,size–;如果head的值等于数组的长度head设置为0;
判断队列是否为空或者满,就需要判断size的值如果size=0,链表为空,在从队列取出元素时,产生堵塞,如果size==数组的长度,那么在放进元素时,也会产生堵塞

下面是代码的实现

class MyBlockingQueue{
    private int[] array = new int[1_0000];
    private int head = 0;
    private int end = 0;
    private int size = 0;
     public void put(int value) throws InterruptedException {
        if(size == array.length){
            return;
        }
        array[end] = value;
        end++;
        //需要处理end下表到达数组末尾的情况
        if(end >= array.length){
            end = 0;
        }
        size++;
        
    }
     public Integer take() throws InterruptedException {
        if(size == 0){
           return null;
        }
        int ret = array[head];
        head++;
        if(head>=array.length){
            head=0;
        }
        size--;
        
        return ret;
    }

}

2. 保证阻塞队列的安全
   从上面的代码我可以看出,take与put方法每一行代码都在对公共变量进行操作,所以我们直接对这两个方法进行加锁即可

class MyBlockingQueue{
    private int[] array = new int[1_0000];
    private int head = 0;
    private int end = 0;
    private int size = 0;
    synchronized public void put(int value) throws InterruptedException {
        if(size == array.length){
            
        }
        array[end] = value;
        end++;
        //需要处理end下表到达数组末尾的情况
        if(end >= array.length){
            end = 0;
        }
        size++;
        
    }
    synchronized public Integer take() throws InterruptedException {
        if(size == 0){
            
        }
        int ret = array[head];
        head++;
        if(head>=array.length){
            head=0;
        }
        size--;
        
        return ret;
    }

}

3. 实现堵塞效果

出队列,head++,size–;如果head的值等于数组的长度head设置为0;
判断队列是否为空或者满,就需要判断size的值如果size=0,链表为空,在从队列取出元素时,产生堵塞,如果size数组的长度,那么在放进元素时,也会产生堵塞

关键使用wait和notify关键字

class MyBlockingQueue{
    private int[] array = new int[1_0000];
    private int head = 0;
    private int end = 0;
    private int size = 0;
    synchronized public void put(int value) throws InterruptedException {
        if(size == array.length){
            this.wait();
        }
        array[end] = value;
        end++;
        //需要处理end下表到达数组末尾的情况
        if(end >= array.length){
            end = 0;
        }
        size++;
        this.notify();
    }
    synchronized public Integer take() throws InterruptedException {
        if(size == 0){
            this.wait();
        }
        int ret = array[head];
        head++;
        if(head>=array.length){
            head=0;
        }
        size--;
        this.notify();
        return ret;
    }

}

这样的一个简单的堵塞队列就完成了
‘下面创建一个实例来感受一下这样的阻塞队列

🌰:`

public class TestDemo3 {
    private static MyBlockingQueue myBlockingQueue = new MyBlockingQueue();

    public static void main1(String[] args) throws InterruptedException {
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>();
        blockingQueue.put("hello");
        blockingQueue.take();
    }
    public static void main(String[] args) {
        Thread producer = new Thread(()->{
            int num = 0;
            while(true){
                try {
                    System.out.println("生产了"+num);
                     myBlockingQueue.put(num);
                    num++;
                    //Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }

        });
        producer.start();
        Thread customer = new Thread(()->{
            while(true){
                try {
                    int num = myBlockingQueue.take();
                    System.out.println("消费了"+num);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        customer.start();

    }
}

运行结果
以上就是阻塞队列的相关基础知识
👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋
👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋
👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋👋

先验知识

  在解释这几个概念之前，需要注意的是：

• 首先需要知道操作系统层面的同步、异步、阻塞这几个概念的含义。关于这方面的内容，可见笔者的另一篇博客：

  同步与异步、并行与并发、阻塞与挂起：
  https://blog.csdn.net/wangpaiblog/article/details/116114098

• 本文解释的概念至少适用于编程语言层面，但不适用于操作系统层面。原因是，在软件工程中，任何设计都可以进行分层封装。其中，每个中间层的设计对上层与下层来说都是透明的。因此，编程语言层面的“同步、异步、阻塞”与操作系统层面的没有必然联系。

• 有人喜欢对“同步阻塞、同步非阻塞、异步阻塞、异步非阻塞”的概念，首先区分发送方和接收方，但实际上，无论是发送还是接收，本质上都是一种“请求”的过程，如果不在操作系统层面进行分析，区分这两者实无必要。

• 同步、异步是针对两个任务（此任务不同于操作系统理论中的作业、进程等，指的只是一个线程中要做的一件事情。本文中的任务都指的是这个。）来说的，其中一个任务为一开始就在执行、所主要关注的，称为主任务，而另一个是与主任务相关的一个任务，称为相关任务。当主任务与相关任务分别位于不同的线程时，称该线程分别为主线程、相关线程。

• 阻塞是针对主任务所在的线程来说的。当主任务和相关任务位于同一个线程时，不存在“阻塞”，即此情况下不存在同步阻塞、异步阻塞。

  但对于下面的情况，主任务和相关任务一定位于不同的线程：

  • 主任务发起的是 I/O 请求

  • 主任务发起的是网络请求

• 请求是一个期望获得资源的行为，而“同步阻塞、同步非阻塞、异步阻塞、异步非阻塞”描述的是获得期望资源之前的行为，而不是其之后的行为。这四个概念中的“阻塞”不描述在获得资源之后主线程的程序走向。

此处的异步指的是什么

  首先需要明白，无论是对“异步”还是“非阻塞”的主线程，都不可能凭空接收到外界、执行没有事先设置的程序。而之所以主线程的行为会受到相关线程的影响，是因为主线程会周期性地调用一种“请求”函数，而此“请求”函数的行为会受相关线程的影响。

  其次，对于操作系统，只要其启动，就一直在运行程序。如果没有显式的进程，就运行一个默认的空进程。另外，编程人员不可能事先预计用户本次对本软件的使用时长，然后在软件中设置一个运行时间与其正好相等的一种任务。这就是说，线程在执行时，实际上是在无限循环、周期性地执行一系列任务。于是，可以在每个任务之间插入一个中断点，用于执行额外的操作（任务），以实现对其它线程的交互，这就是异步操作。理由是，中断点之前的任务都已经完成，而任务之间本是不应有任务的（因为任务之间的部分属于空档，没有任何东西），因此，任务之间的中断点是由中断点之前的某任务插入的。由于某种原因，该任务并不想马上执行某操作，而是选择将其移至一系列任务执行完之后的某个任务之间的空档来执行，因此称为异步操作。

同步、异步、阻塞、非阻塞

  前面已经指出了关键性的知识点，下面将直接给“同步阻塞、同步非阻塞、异步阻塞、异步非阻塞”的概念。为了更好的说明，笔者做了一张图，如下：

  对于主线程上执行的一系列任务，如果其中的某个任务需要与相关线程交互，当其立即暂时当前的任务而发起请求，这称为同步；当该任务将请求的时间安排到某任务完成之后再发起，这称为异步。当发起请求时，直至获得完整的资源之后，不会继续执行现在或之后的任务，这称为阻塞；当发起请求时，立即获得瞬时的结果，然后继续执行现在或之后的任务，如果获得的瞬时资源不是完整的资源，将之后周期性发送类似的请求，直至获得完整的资源，这称为非阻塞。

  可以看出，同步与异步的区别在于发起请求的时机，而阻塞与非阻塞的区别在于发起请求后是否对本线程进行暂停。

同步阻塞、同步非阻塞、异步阻塞、异步非阻塞

  很多读者（包含笔者）都喜欢作者能直截了当地给出概念，而反感拐弯抹角和旁敲侧击。因此，笔者再提炼一下本文核心的四个概念：

• 同步阻塞：在需要某资源时马上发起请求，并暂停本线程之后的程序，直至获得所需的资源。

• 同步非阻塞：在需要某资源时马上发起请求，且可以马上得到答复，然后继续执行之后的程序。但如果得到的不是完整的资源，之后将周期性地的请求。

• 异步阻塞：在需要某资源时不马上发起请求，而安排一个以后的时间再发起请求。当到了那时发出请求时，将暂停本线程之后的程序，直至获得所需的资源。

• 异步非阻塞：在需要某资源时不马上发起请求，而安排一个以后的时间再发起请求。当到了那时发出请求时，可以马上得到答复，然后继续执行之后的程序。但如果得到的不是完整的资源，之后将周期性地的请求。

【提醒】

  有些过分钻研概念的极客可能会对此提出质疑：既然“不马上发起请求”属于异步，但是如果一个线程将 CPU 分配的时间片用完了，此时应该得不到执行，那么同步非阻塞是不是可以归于异步非阻塞呢？

  实际上不是这样。一方面，这里的“马上执行”，指的是针对这个程序中相关的代码而言的。基于上一条代码而执行的下一条代码就称为“被马上执行的代码”，而不管这两条代码的执行之间有没有被 CPU 中断。另一方面，评判 同步非阻塞 与 异步阻塞 的其中一个标准是，如果一个主线程中，依次调用了 N 个 同步非阻塞 的方法，则这些方法被执行的次序是不确定的。但如果一个主线程中，依次调用了 N 个 异步阻塞 的方法，则当主线程完成当前的任务之后，转而真正执行这些异步方法时，这些方法被执行的次序与之前调用时一致。

一个生动的例子

  一个贴切生动的例子可能对理解更有帮助。这里假设了这样的一种情景：笔者正在进行公司安排的一个“cleancode”专项需求（下面简称 cleancode 专项），然后突然对于门禁上报告的一项告警不太理解，笔者想要求助自己的同事（设该同事名为 Bob），于是在公司的通信软件（设该软件名为 contact）上向其发送了此求助消息，并假设笔者每天有减脂的诉求，因此在下午下班后不会立刻去吃饭。

  前述的四个概念类比如下：

• 同步阻塞：笔者在 contact 上给 Bob 发了一条咨询信息，并开启 contact 的消息自动弹出功能。然后笔者暂停手头的工作，翘着二郎腿开始用手机摸鱼，直到手机上弹出 contact 的关于 Bob 的回复信息。

  解释：

  • 笔者、笔者的同事 Bob：两个位于不同服务器上的操作系统。

  • cleancode 专项：正在“笔者操作系统”上运行的一个线程。

  • 笔者放弃工作上的任务：cleancode 专项线程被阻塞。

  • 笔者开始摸鱼：cleancode 专项线程因阻塞而使笔者空闲，笔者通过调度来运行其它无关线程。

  • contact 的消息自动弹出：在操作系统中，用于唤醒阻塞线程的信号量。

• 同步非阻塞：笔者在 contact 上给 Bob 发了一条咨询信息，然后笔者继续做 cleancode 专项中的其它内容，并周期性查看 Bob 有没有回复。

• 异步阻塞：笔者在日程表上记录了这个待办事项，然后笔者继续做 cleancode 专项中的其它内容，最后到下午下班时，笔者在 contact 上给 Bob 发了一条咨询信息，并开启 contact 的消息自动弹出功能。然后笔者暂停手头的工作，翘着二郎腿开始在晚上加班时间用手机摸鱼，直到手机上弹出 contact 的关于 Bob 的回复信息。

• 异步非阻塞：笔者在日程表上记录了这个待办事项，然后笔者继续做 cleancode 专项中的其它内容，最后到下午下班时，笔者在 contact 上给 Bob 发了一条咨询信息，然后笔者继续加班做 cleancode 专项中的其它内容，并周期性查看 Bob 有没有回复。

总结与补充

• 用一句话概括本文的概念：

  • 异步：把事情推到以后去做

  • 阻塞：专心做一件事情

  • 同步非阻塞：一边做一件事情，一边做另一件事情（一心二用）

  • 异步阻塞：把问题推到以后专心处理

  • 同步阻塞：马上专心做一件事情

  • 异步非阻塞：把问题推到以后时不时处理一下

• 线程在被阻塞时，CPU 会将时间片分给其它线程。而当线程发出非阻塞线程请求时，它以后还要周期性地请求，这同样会占用 CPU 时间。因此，不能一味地认为异步非阻塞一定优于同步非阻塞。同样是异步非阻塞，底层实现不同，效率也不同。将程序中的所有代码都改成异步非阻塞，也未必可以提高系统的整体性能。哪种性能最好要取决于具体实现，而不是几个用于装蒜的术语。

• Java 中没有异步关键字，所以一般情况下，Java 代码都是同步的，Java 中只有同步阻塞和同步非阻塞。但异步代码可以通过同步代码设计出来，所以 Java 中也可以设计出异步方法。

  JavaScript 有异步关键字，但 JavaScript 是单线程的，所以 JavaScript 中只有同步阻塞和异步阻塞。

• 我们平常单独使用的“同步”一词，实际上指的是这里的同步阻塞，而“异步”指异步非阻塞。

• 在操作系统层面，只有单独的同步与异步、阻塞与非阻塞的说法。此时的同步与异步的含义主要有以下几个：

  • 同本文的同步阻塞、异步非阻塞。

  • 同步：强调两个程序的运行彼此有逻辑、时间上的先后关系。

    异步：强调两个程序的运行彼此相对独立。