前言
在去年的面试过程中，被面试官问道“阻塞队列”这个问题，因为过后并没有对此问题进行深刻了解，只是依照本人的了解阐明了该问题，最初面试后果也不太好，明天对该问题进行简要的面试并记录如下；如有谬误，欢送斧正。
什么是阻塞队列
在数据结构中，队列遵循FIFO（先进先出）准则。在java中，Queue接口定义了定义了根本行为，由子类实现实现，常见的队列有ArrayDeque、LinkedList等，这些都是非线程平安的，在java 1.5中新增了阻塞队列，当队列满时，增加元素的线程呈阻塞状态；当队列为空时，获取元素的线程呈阻塞状态。
生产者、消费者模型
生产者将元素增加到队列中，生产中获取数据后实现数据处理。两者通过队列解决了生产者和消费者的耦合关系；当生产者的生产速度与消费者的生产速度不统一时，能够通过小道缓冲的目标。
阻塞队列的应用场景
线程池
在线程池中，当工作线程数大于等于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
- add(e):当队列已满时，再增加元素会抛出异样
删除元素：
- remove(e):返回true示意已胜利删除，否则返回false
- poll():如果队列为空返回null，否则返回队列中的第一个元素
- take():获取队列中的第一个元素，如果队列为空，获取元素的线程变为阻塞状态
- poll(time, unit)：当队列为空时，线程被阻塞，如果超过指定工夫，线程退出
查看元素：
- element():获取队头元素，如果元素为null，抛出
NoSuchElementException - peek():获取队头元素，如果队列为空返回null，否则返回指标元素
- element():获取队头元素，如果元素为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的准则解决，反之；默认为false
- c:已有元素的汇合，类型于合并两个数组
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()和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)
浏览原文