自己动手实现一个阻塞队列

约定不等于承诺〃 2023-10-04 11:29 57阅读 0赞

# 1. 阻塞队列介绍 #

顾名思义，阻塞队列是一个具备先进先出特性的队列结构，从队列末尾插入数据，从队列头部取出数据。而阻塞队列与普通队列的最大不同在于阻塞队列提供了阻塞式的同步插入、取出数据的功能（阻塞入队put/阻塞出队take）。

使用**put**插入数据时，如果队列空间已满并不直接返回，而是令当前操作的线程陷入阻塞态（生产者线程），等待着阻塞队列中的元素被其它线程（消费者线程）取走，令队列重新变得不满时被唤醒再次尝试插入数据。使用**take**取出数据时，如果队列空间为空并不直接返回，而是令当前操作的线程陷入阻塞态（消费者线程），等待其它线程（生产者线程）插入新元素，令队列非空时被唤醒再次尝试取出数据。

阻塞队列主要用于解决并发场景下消费者线程与生产者线程处理速度不一致的问题。例如jdk的线程池实现中，线程池核心线程（消费者线程）处理速度一定的情况下，如果业务方线程提交的任务过多导致核心线程处理不过来时，将任务暂时放进阻塞队列等待核心线程消费（阻塞队列未满）；由于核心线程常驻的原因，当业务方线程提交的任务较少，核心线程消费速度高于业务方生产速度时，核心线程作为消费者会阻塞在阻塞队列的take方法中，避免无谓的浪费cpu资源。

由于阻塞队列在内部实现了协调生产者/消费者的机制而不需要外部使用者过多的考虑并发同步问题，极大的降低了生产者/消费者场景下程序的复杂度。

# 2. 自己实现阻塞队列 #

下面我们自己动手一步步的实现几个不同版本、效率由低到高的的阻塞队列，来加深对阻塞队列工作原理的理解。

## 阻塞队列接口 ##

为了降低复杂度，我们的阻塞队列只提供最基础的出队、入队和判空接口。

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

/**
     * 阻塞队列
     * 1. 首先是一个先进先出的队列
     * 2. 提供特别的api，在入队时如果队列已满令当前操作线程阻塞；在出队时如果队列为空令当前操作线程阻塞
     * 3. 单个元素的插入、删除操作是线程安全的
     */
    public interface MyBlockingQueue<E> {
    
       /**
         * 插入特定元素e，加入队尾
         * 队列已满时阻塞当前线程，直到队列中元素被其它线程删除并插入成功
         * */
        void put(E e) throws InterruptedException;
    
        /**
         * 队列头部的元素出队(返回头部元素，将其从队列中删除)
         * 队列为空时阻塞当前线程，直到队列被其它元素插入新元素并出队成功
         * */
        E take() throws InterruptedException;
    
        /**
         * 队列是否为空
         * */
        boolean isEmpty();
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

## 2.1 v1版本（最基本的队列实现） ##

博客中所实现的阻塞队列底层是使用数组承载数据的（ArrayBlockingQueue），内部提供了私有方法enqueue和dequeue来实现原始的内部入队和出队操作。

最初始的v1版本中，我们只实现最基本的FIFO队列功能，其put和take方法只是简单的调用了enqueue和dequeue，因此v1版本中其入队、出队不是阻塞的，也无法保障线程安全，十分简陋。

**后续的版本中，我们会以v1版本为基础，实现阻塞调用以及线程安全的特性，并且对所实现的阻塞队列性能进行不断的优化。**

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

/**
     * 数组作为底层结构的阻塞队列 v1版本
     */
    public class MyArrayBlockingQueueV1<E> implements MyBlockingQueue<E> {
    
        /**
         * 队列默认的容量大小
         * */
        private static final int DEFAULT_CAPACITY = 16;
    
        /**
         * 承载队列元素的底层数组
         * */
        private final Object[] elements;
    
        /**
         * 当前头部元素的下标
         * */
        private int head;
    
        /**
         * 下一个元素插入时的下标
         * */
        private int tail;
    
        /**
         * 队列中元素个数
         * */
        private int count;
    
        //=================================================构造方法======================================================
        public MyArrayBlockingQueueV1() {
            // 设置数组大小为默认
            this.elements = new Object[DEFAULT_CAPACITY];
    
            // 初始化队列 头部,尾部下标
            this.head = 0;
            this.tail = 0;
        }
    
        public MyArrayBlockingQueueV1(int initCapacity) {
            assert initCapacity > 0;
    
            this.elements = new Object[initCapacity];
    
            // 初始化队列 头部,尾部下标
            this.head = 0;
            this.tail = 0;
        }
    
        /**
         * 下标取模
         * */
        private int getMod(int logicIndex){
            int innerArrayLength = this.elements.length;
    
            // 由于队列下标逻辑上是循环的
            if(logicIndex < 0){
                // 当逻辑下标小于零时
    
                // 真实下标 = 逻辑下标 + 加上当前数组长度
                return logicIndex + innerArrayLength;
            } else if(logicIndex >= innerArrayLength){
                // 当逻辑下标大于数组长度时
    
                // 真实下标 = 逻辑下标 - 减去当前数组长度
                return logicIndex - innerArrayLength;
            } else {
                // 真实下标 = 逻辑下标
                return logicIndex;
            }
        }
    
        /**
         * 入队
         * */
        private void enqueue(E e){
            // 存放新插入的元素
            this.elements[this.tail] = e;
            // 尾部插入新元素后 tail下标后移一位
            this.tail = getMod(this.tail + 1);
    
            this.count++;
        }
    
        /**
         * 出队
         * */
        private E dequeue(){
            // 暂存需要被删除的数据
            E dataNeedRemove = (E)this.elements[this.head];
            // 将当前头部元素引用释放
            this.elements[this.head] = null;
            // 头部下标 后移一位
            this.head = getMod(this.head + 1);
    
            this.count--;
    
            return dataNeedRemove;
        }
    
        @Override
        public void put(E e){
            enqueue(e);
        }
    
        @Override
        public E take() {
           return dequeue();
        }
    
        @Override
        public boolean isEmpty() {
            return this.count == 0;
        }
    }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

## 2.2 v2版本（实现同步阻塞和线程安全的特性） ##

前面提到阻塞调用的出队、入队的功能是阻塞队列区别于普通队列的关键特性。阻塞调用实现的方式有很多，其中最容易理解的一种方式便是无限循环的轮询，直到出队/入队成功（虽然cpu效率很低）。

**v2版本在v1的基础上，使用无限循环加定时休眠的方式简单的实现了同步调用时阻塞的特性。****并且在put/take内增加了synchronized块将入队/出队代码包裹起来，阻止多个线程并发的操作队列而产生线程安全问题。**

**v2版本入队方法实现：**

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

@Override
        public void put(E e) throws InterruptedException {
            while (true) {
                synchronized (this) {
                    // 队列未满时执行入队操作
                    if (count != elements.length) {
                        // 入队，并返回
                        enqueue(e);
                        return;
                    }
                }
    
                // 队列已满，休眠一段时间后重试
                Thread.sleep(100L);
            }
        }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

**v2版本出队方法实现：**

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

@Override
        public E take() throws InterruptedException {
            while (true) {
                synchronized (this) {
                    // 队列非空时执行出队操作
                    if (count != 0) {
                        // 出队并立即返回
                        return dequeue();
                    }
                }
    
                // 队列为空的情况下,休眠一段时间后重试
                Thread.sleep(100L);
            }
        }

![复制代码][48304ba5e6f9fe08f3fa1abda7d326ab.png]

**v2版本完整代码：**

![8f900a89c6347c561fdf2122f13be562.png][]

## 2.3 v3版本（引入条件变量优化无限循环轮询） ##

在有大量线程竞争的情况下，v2版本无限循环加休眠的阻塞方式存在两个严重的问题。

### 无限循环轮询的缺陷 ###

**1. 线程周期性的休眠/唤醒会造成频繁的发生线程上下文切换，非常浪费cpu资源**</

[48304ba5e6f9fe08f3fa1abda7d326ab.png]: https://img-blog.csdnimg.cn/img_convert/48304ba5e6f9fe08f3fa1abda7d326ab.png
[8f900a89c6347c561fdf2122f13be562.png]: https://img-blog.csdnimg.cn/img_convert/8f900a89c6347c561fdf2122f13be562.png

发表评论取消回复

表情：

评论列表（有 0 条评论，57人围观）

还没有评论，来说两句吧...

相关阅读

相关从零开始自己动手写阻塞队列

![038e1acc0733485a85601975dd73d3c6.png][] 前言在我们平时编程的时候一个很重要的工具就是容器，在本篇文章当中主要给大家介绍阻塞

小咪咪/ 2024年04月08日 13:52/ 0 赞/ 82 阅读

相关自己动手实现一个阻塞队列

1. 阻塞队列介绍顾名思义，阻塞队列是一个具备先进先出特性的队列结构，从队列末尾插入数据，从队列头部取出数据。而阻塞队列与普通队列的最大不同在于阻塞队列提供了阻塞式的同

约定不等于承诺〃/ 2023年10月04日 11:29/ 0 赞/ 58 阅读

相关 C++ 实现一个简易的阻塞队列

阻塞队列是多线程中常用的数据结构，对于实现多线程之间的数据交换、同步等有很大作用。阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素

川长思鸟来/ 2022年12月11日 15:11/ 0 赞/ 153 阅读

相关自己动手实现一个malloc内存分配器 | 30图

如果你不能理解malloc之类内存分配器实现原理的话，那你可能写不出高性能程序，写不出高性能程序就很难参与核心项目，参与不了核心项目那么很难升职加薪，很难升级加薪就无法走向人生

浅浅的花香味﹌/ 2022年11月11日 14:44/ 0 赞/ 195 阅读

相关自己动手实现消息队列之JMS

什么是JMS?JMS的诞生史? 在JMS还没有诞生前，每个企业都会有自己的一套内部消息系统，比如项目组A需要调用到项目组B的系统，项目组B也有可能会调用到项目组C的系统。

川长思鸟来/ 2022年07月18日 00:13/ 0 赞/ 207 阅读

相关自己动手实现一个简单的 IOC

再上一篇文章中，楼主和大家一起分析spring的 IOC 实现，剖析了Spring的源码，看的出来，源码异常复杂，这是因为Spring的设计者需要考虑到框架的扩展性，健壮性，性

痛定思痛。/ 2022年06月04日 05:37/ 0 赞/ 214 阅读

相关自己动手实现一个简单的 SpringMVC

自己动手实现一个SpringMvc springmvc 作为javaweb 主流框架,相比大家都用过,也已经都很熟悉了.但是对springmvc如何实现有了解多少呢?

超、凢脫俗/ 2022年05月28日 12:56/ 0 赞/ 247 阅读

相关自己动手实现一个MVVM库

[自己动手实现一个MVVM库][MVVM] 我们知道的，常见的数据绑定的实现方法 1、数据劫持(vue)：通过[Object.defineProperty()][Obj

小鱼儿/ 2022年03月12日 15:50/ 0 赞/ 274 阅读

相关如何自己实现一个队列

前言队列是一种先进先出的数据结构，也是常见的数据结构之一。日常生活中的排队买东西就是一种典型的队列，而在购票系统也需要一个队列处理用户的购票请求，当然这里的队列就复杂多

水深无声/ 2022年02月25日 15:38/ 0 赞/ 278 阅读

相关自己动手实现一个Android Studio插件

在使用Android Studio开发的时候，大部分人都会使用一些插件来提高开发效率，例如我们所熟知的butternife，selector,，GsonFormat等，这些分别

喜欢ヅ旅行/ 2022年01月27日 14:19/ 0 赞/ 389 阅读