详解Java中的阻塞队列

 更新时间:2021年06月16日 14:49:33   作者:小海子l  
在去年的面试过程中,被面试官问道“阻塞队列”这个问题,因为当时并没有对此问题进行深入理解,只是按照自己的理解说明了该问题,最后面试结果也不太好,今天对该问题进行简要的面试并记录如下;如有错误,欢迎指正,需要的朋友可以参考下

什么是阻塞队列

在数据结构中,队列遵循FIFO(先进先出)原则。在java中,Queue接口定义了定义了基本行为,由子类完成实现,常见的队列有ArrayDequeLinkedList等,这些都是非线程安全的,在java 1.5中新增了阻塞队列,当队列满时,添加元素的线程呈阻塞状态;当队列为空时,获取元素的线程呈阻塞状态。

生产者、消费者模型

在这里插入图片描述

生产者将元素添加到队列中,消费中获取数据后完成数据处理。两者通过队列解决了生产者和消费者的耦合关系;当生产者的生产速度与消费者的消费速度不一致时,可以通过大道缓冲的目的。

阻塞队列的使用场景

线程池

在线程池中,当工作线程数大于等于corePoolSize时,后续的任务后添加到阻塞队列中;

目前有那些阻塞队列

在java中,BlockingQueue接口定义了阻塞队列的行为,常用子类是ArrayBlockingQueueLinkedBlockingQueue

在这里插入图片描述

BlockingQueue继承了Queue接口,拥有其全部特性。在BlockingQueue的java doc中对其中的操作方法做了汇总

在这里插入图片描述

插入元素

  • add(e):当队列已满时,再添加元素会抛出异常IllegalStateException
  • offer(e):添加成功,返回true,否则返回false
  • put:(e):当队列已满时,再添加元素会使线程变为阻塞状态
  • offer(e, time,unit):当队列已满时,在末尾添加数据,如果在指定时间内没有添加成功,返回false,反之是true

删除元素

  •  remove(e):返回true表示已成功删除,否则返回false
  • poll():如果队列为空返回null,否则返回队列中的第一个元素
  • take():获取队列中的第一个元素,如果队列为空,获取元素的线程变为阻塞状态
  • poll(time, unit):当队列为空时,线程被阻塞,如果超过指定时间,线程退出

检查元素

  •  element():获取队头元素,如果元素为null,抛出NoSuchElementException
  • peek():获取队头元素,如果队列为空返回null,否则返回目标元素

ArrayBlockingQueue

底层基于数组的有界阻塞队列,在构造此队列时必须指定容量;

构造函数

// 第一个	
public ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c) {
        this(capacity, fair);

        final ReentrantLock lock = this.lock;
        lock.lock(); // Lock only for visibility, not mutual exclusion
        try {
            int i = 0;
            try {
                for (E e : c) {
                    checkNotNull(e);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
                throw new IllegalArgumentException();
            }
            count = i;
            putIndex = (i == capacity) ? 0 : i;
        } finally {
            lock.unlock();
        }
    }

	// 第二个
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

	// 第三个
	public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }
  • capacity:队列的初始容量
  • fair:线程访问队列的公平性。如果为true按照FIFO的原则处理,反之;默认为falsec:
  • 已有元素的集合,类型于合并两个数组

put()方法

public void put(E e) throws InterruptedException {
         // 检查元素是否为null
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        // 获取锁
        lock.lockInterruptibly();
        try {
            // 如果当前队列为空,变为阻塞状态
            while (count == items.length)
                notFull.await();
            // 反之,就添加元素
            enqueue(e);
        } finally {
            // 解锁
            lock.unlock();
        }
    }

    private void enqueue(E x) {
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        // 此时队列不为空,唤醒消费者
        notEmpty.signal();
    }

take()方法

 public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        // 获取锁
        lock.lockInterruptibly();
        try {
            // 如果队列为空,消费者变为阻塞状态
            while (count == 0)
                notEmpty.await();
            // 不为空,就获取数据
            return dequeue();
        } finally {
            // 解锁
            lock.unlock();
        }
    }

        private E dequeue() {
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        // 获取队头元素x
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
         // 此时队列没有满,同时生产者继续添加数据
        notFull.signal();
        return x;
    }

LinkedBlockingQueue

底层基于单向链表的无界阻塞队列,如果不指定初始容量,默认为Integer.MAX_VALUE,否则为指定容量

构造函数

// 不指定容量     
	public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }
	// 指定容量
    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

	// 等同于合并数组
    public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock(); // Never contended, but necessary for visibility
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

put()方法

 public void put(E e) throws InterruptedException {
        // 元素为空,抛出异常
        if (e == null) throw new NullPointerException();
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        // 获取队列中的数据量
        final AtomicInteger count = this.count;
        // 获取锁
        putLock.lockInterruptibly();
        try {
            // 队列满了,变为阻塞状态
            while (count.get() == capacity) {
                notFull.await();
            }
            // 将目标元素添加到链表的尾端
            enqueue(node);
            // 总数增加
            c = count.getAndIncrement();
            // 队列还没有满,继续添加元素
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            // 解锁
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

take()方法

public E take() throws InterruptedException {
        E x;
        int c = -1;
        // 获取队列中的工作数
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        // 获取锁
        takeLock.lockInterruptibly();
        try {
            // 如果队列为空,变为阻塞状态
            while (count.get() == 0) {
                notEmpty.await();
            }
            // 获取队头元素
            x = dequeue();
            // 递减
            c = count.getAndDecrement();
            // 通知消费者
            if (c > 1)
                notEmpty.signal();
        } finally {
            // 解锁
            takeLock.unlock();
        }
        if (c == capacity)
            // 
            signalNotFull();
        return x;
    }

对比

相同点

  • 两者都是通过Condition通知生产者和消费者完成元素的添加和获取
  • 都可以指定容量

不同点

  •  ArrayBlockingQueue基于数据,LinkedBlockingQueue基于链表
  • ArrayBlockingQueue内有一把锁,LinkedBlockingQueue内有两把锁

在这里插入图片描述
在这里插入图片描述 

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

通过分析源码可以知道,阻塞队列其实是通过通知机制Condition完成生产者和消费的互通。也可以通过Object类中的wait()notifynotifyAll实现。下面是自己写的一个阻塞队列

public class BlockQueue {
    // 对象锁
    public static final Object LOCK = new Object();
    // 控制变量的值 来通知双方
    public boolean condition;
    
    public void put() {
        synchronized (LOCK) {
            while (condition) {
                try {
                    // 满了
                    System.out.println("put   队列满了,开始阻塞");
                    LOCK.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            condition = true;
            System.out.println("put   改为true,唤醒消费者");
            LOCK.notifyAll();
        }
    }


    public void take() {
        synchronized (LOCK) {
            while (!condition) {
                // 没满
                System.out.println("take   队列没满,开始阻塞");
                try {
                    LOCK.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            condition = false;
            System.out.println("take   改为false,唤醒生产者");
            LOCK.notifyAll();
        }
    }
}

参考文章:

并发容器之BlockingQueue (juejin.cn)

BlockingQueue (Java Platform SE 8 ) (oracle.com)

到此这篇关于详解Java中的阻塞队列的文章就介绍到这了,更多相关Java阻塞队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • SpringBoot2.x 集成 Thymeleaf的详细教程

    SpringBoot2.x 集成 Thymeleaf的详细教程

    本文主要对SpringBoot2.x集成Thymeleaf及其常用语法进行简单总结,其中SpringBoot使用的2.4.5版本。对SpringBoot2.x 集成 Thymeleaf知识感兴趣的朋友跟随小编一起看看吧
    2021-07-07
  • 一文带你搞懂如何在Spring Boot中轻松调用Python脚本

    一文带你搞懂如何在Spring Boot中轻松调用Python脚本

    虽然springBoot非常适合开发高性能的Web应用,但它的生态中并没有像Python那样丰富的科学计算库,所以最理想的方案就是将Spring Boot和Python结合,下面我们来看看具体的实现方法吧
    2025-04-04
  • Springboot微服务打包Docker镜像流程解析

    Springboot微服务打包Docker镜像流程解析

    这篇文章主要介绍了Springboot微服务打包Docker镜像流程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-08-08
  • SpringBoot异常处理机制使用详解

    SpringBoot异常处理机制使用详解

    SpringBoot异常处理基于Servlet与Spring框架,通过ErrorAttributes、ErrorViewResolver、BasicErrorController协作实现统一异常响应,企业级应用需标准化异常体系,统一响应格式,优化性能与安全,如隐藏堆栈信息,确保日志上下文
    2025-08-08
  • springboot中配置项目url的常见方法详解

    springboot中配置项目url的常见方法详解

    这篇文章主要为大家详细介绍了springboot中配置项目url的两个常见方法,文中的示例代码讲解详细,具有一定的借鉴价值,有需要的小伙伴可以了解下
    2025-11-11
  • Java从控制台接受输入字符的简单方法

    Java从控制台接受输入字符的简单方法

    这篇文章主要介绍了Java从控制台接受输入字符的简单方法,需要的朋友可以参考下
    2014-02-02
  • SpringBoot整合Swagger接口文档工具的流程步骤

    SpringBoot整合Swagger接口文档工具的流程步骤

    我们在开发接口的时候,会将接口文档给前端的开发者进行对接,我们可以通过Postman或者Yapi等接口管理工具进行编写管理,实际开发中,接口的管理确实也应该通过专业的工具管理,本文,我们就来谈谈怎么在SpringBoot整合Swagger接口文档工具
    2023-08-08
  • SpringBoot 应用程序测试实现方案

    SpringBoot 应用程序测试实现方案

    这篇文章主要介绍了SpringBoot 应用程序测试实现方案,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-11-11
  • java 导入Excel思路及代码示例

    java 导入Excel思路及代码示例

    这篇文章主要介绍了java 导入Excel思路及代码示例,简要分析了设计思路,然后分享了实例代码,具有一定参考价值,需要的朋友可以了解下。
    2017-11-11
  • MyBatis-Plus多表联合查询并且分页(3表联合)

    MyBatis-Plus多表联合查询并且分页(3表联合)

    这篇文章主要介绍了MyBatis-Plus多表联合查询并且分页(3表联合),文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2020-08-08

最新评论