详解Java中的阻塞队列详解Java中的阻塞队列

什么是阻塞队列在数据结构中，队列遵循FIFO（先进先出）原则。在java中，Queue接口定义了定义了基本行为，由子类完成实现，常见的队列有ArrayDeque、LinkedList等，这些都是非线程安全的，在java 1.5中新增了阻塞队列，当队列满时，添加元素的线程呈阻塞状态；当队列为空时，获取元素的线程呈阻塞状态。
生产者、消费者模型 【详解Java中的阻塞队列】
文章图片

生产者将元素添加到队列中，消费中获取数据后完成数据处理。两者通过队列解决了生产者和消费者的耦合关系；当生产者的生产速度与消费者的消费速度不一致时，可以通过大道缓冲的目的。
阻塞队列的使用场景线程池
在线程池中，当工作线程数大于等于corePoolSize时，后续的任务后添加到阻塞队列中；
目前有那些阻塞队列在java中，BlockingQueue接口定义了阻塞队列的行为，常用子类是ArrayBlockingQueue和LinkedBlockingQueue。

文章图片

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 c) {this(capacity, fair); final ReentrantLock lock = this.lock; lock.lock(); // Lock only for visibility, not mutual exclusiontry {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 {// 检查元素是否为nullcheckNotNull(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")// 获取队头元素xE 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(null); } // 等同于合并数组public LinkedBlockingQueue(Collection c) {this(Integer.MAX_VALUE); final ReentrantLock putLock = this.putLock; putLock.lock(); // Never contended, but necessary for visibilitytry {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)); ++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 node = new Node(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()和notify、notifyAll实现。下面是自己写的一个阻塞队列

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阻塞队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！