前言
在去年的面试过程中，被面试官问道“阻塞队列”这个问题，因为过后并没有对此问题进行深刻了解，只是依照本人的了解阐明了该问题，最初面试后果也不太好，明天对该问题进行简要的面试并记录如下；如有谬误，欢送斧正。
什么是阻塞队列
在数据结构中，队列遵循 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)
浏览原文