用Java自己动手实现一个阻塞队列
共 75076字，需浏览 151分钟
·
2021-04-25 10:04
点击上方蓝色字体，选择“标星公众号”
优质文章，第一时间送达
作者 | 小熊餐馆
来源 | urlify.cn/IbIB7j
1. 阻塞队列介绍
顾名思义，阻塞队列是一个具备先进先出特性的队列结构，从队列末尾插入数据，从队列头部取出数据。而阻塞队列与普通队列的最大不同在于阻塞队列提供了阻塞式的同步插入、取出数据的功能（阻塞入队put/阻塞出队take）。
使用put插入数据时，如果队列空间已满并不直接返回，而是令当前操作的线程陷入阻塞态（生产者线程），等待着阻塞队列中的元素被其它线程（消费者线程）取走，令队列重新变得不满时被唤醒再次尝试插入数据。使用take取出数据时，如果队列空间为空并不直接返回，而是令当前操作的线程陷入阻塞态（消费者线程），等待其它线程（生产者线程）插入新元素，令队列非空时被唤醒再次尝试取出数据。
阻塞队列主要用于解决并发场景下消费者线程与生产者线程处理速度不一致的问题。例如jdk的线程池实现中，线程池核心线程（消费者线程）处理速度一定的情况下，如果业务方线程提交的任务过多导致核心线程处理不过来时，将任务暂时放进阻塞队列等待核心线程消费（阻塞队列未满）；由于核心线程常驻的原因，当业务方线程提交的任务较少，核心线程消费速度高于业务方生产速度时，核心线程作为消费者会阻塞在阻塞队列的take方法中，避免无谓的浪费cpu资源。
由于阻塞队列在内部实现了协调生产者/消费者的机制而不需要外部使用者过多的考虑并发同步问题，极大的降低了生产者/消费者场景下程序的复杂度。
2. 自己实现阻塞队列
下面我们自己动手一步步的实现几个不同版本、效率由低到高的的阻塞队列，来加深对阻塞队列工作原理的理解。
阻塞队列接口
为了降低复杂度，我们的阻塞队列只提供最基础的出队、入队和判空接口。
/**
* 阻塞队列
* 1. 首先是一个先进先出的队列
* 2. 提供特别的api，在入队时如果队列已满令当前操作线程阻塞；在出队时如果队列为空令当前操作线程阻塞
* 3. 单个元素的插入、删除操作是线程安全的
*/
public interface MyBlockingQueue<E> {
/**
* 插入特定元素e，加入队尾
* 队列已满时阻塞当前线程，直到队列中元素被其它线程删除并插入成功
* */
void put(E e) throws InterruptedException;
/**
* 队列头部的元素出队(返回头部元素，将其从队列中删除)
* 队列为空时阻塞当前线程，直到队列被其它元素插入新元素并出队成功
* */
E take() throws InterruptedException;
/**
* 队列是否为空
* */
boolean isEmpty();
}
2.1 v1版本（最基本的队列实现）
博客中所实现的阻塞队列底层是使用数组承载数据的（ArrayBlockingQueue），内部提供了私有方法enqueue和dequeue来实现原始的内部入队和出队操作。
最初始的v1版本中，我们只实现最基本的FIFO队列功能，其put和take方法只是简单的调用了enqueue和dequeue，因此v1版本中其入队、出队不是阻塞的，也无法保障线程安全，十分简陋。
后续的版本中，我们会以v1版本为基础，实现阻塞调用以及线程安全的特性，并且对所实现的阻塞队列性能进行不断的优化。
/**
* 数组作为底层结构的阻塞队列 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;
}
}
2.2 v2版本（实现同步阻塞和线程安全的特性）
前面提到阻塞调用的出队、入队的功能是阻塞队列区别于普通队列的关键特性。阻塞调用实现的方式有很多，其中最容易理解的一种方式便是无限循环的轮询，直到出队/入队成功（虽然cpu效率很低）。
v2版本在v1的基础上，使用无限循环加定时休眠的方式简单的实现了同步调用时阻塞的特性。并且在put/take内增加了synchronized块将入队/出队代码包裹起来，阻止多个线程并发的操作队列而产生线程安全问题。
v2版本入队方法实现：
@Override
public void put(E e) throws InterruptedException {
while (true) {
synchronized (this) {
// 队列未满时执行入队操作
if (count != elements.length) {
// 入队，并返回
enqueue(e);
return;
}
}
// 队列已满，休眠一段时间后重试
Thread.sleep(100L);
}
}
v2版本出队方法实现：
@Override
public E take() throws InterruptedException {
while (true) {
synchronized (this) {
// 队列非空时执行出队操作
if (count != 0) {
// 出队并立即返回
return dequeue();
}
}
// 队列为空的情况下,休眠一段时间后重试
Thread.sleep(100L);
}
}
v2版本完整代码：
/**
* 数组作为底层结构的阻塞队列 v2版本
*/
public class MyArrayBlockingQueueV2<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 MyArrayBlockingQueueV2() {
// 设置数组大小为默认
this.elements = new Object[DEFAULT_CAPACITY];
// 初始化队列 头部,尾部下标
this.head = 0;
this.tail = 0;
}
/**
* 默认构造方法
* */
public MyArrayBlockingQueueV2(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) throws InterruptedException {
while (true) {
synchronized (this) {
// 队列未满时执行入队操作
if (count != elements.length) {
// 入队，并返回
enqueue(e);
return;
}
}
// 队列已满，休眠一段时间后重试
Thread.sleep(100L);
}
}
@Override
public E take() throws InterruptedException {
while (true) {
synchronized (this) {
// 队列非空时执行出队操作
if (count != 0) {
// 出队并立即返回
return dequeue();
}
}
// 队列为空的情况下,休眠一段时间后重试
Thread.sleep(100L);
}
}
@Override
public boolean isEmpty() {
return this.count == 0;
}
}
2.3 v3版本（引入条件变量优化无限循环轮询）
在有大量线程竞争的情况下，v2版本无限循环加休眠的阻塞方式存在两个严重的问题。
无限循环轮询的缺陷
1. 线程周期性的休眠/唤醒会造成频繁的发生线程上下文切换，非常浪费cpu资源
2. 线程在尝试操作失败被阻塞时（尝试入队时队列已满、尝试出队时队列为空），如果休眠时间设置的太短，则休眠/唤醒的次数会非常多，cpu性能低下；但如果休眠的时间设置的较长，则会导致被阻塞线程在队列状态发生变化时无法及时的响应
举个例子：某一生产者线程在入队时发现队列已满，当前线程休眠1s，在0.1s之后一个消费者线程取走了一个元素，而此时休眠的生产者线程还需要白白等待0.9s后才被唤醒并感知到队列未满而接着执行入队操作。综上所述，无限循环加休眠的v2版本阻塞队列其性能极差，需要进一步的优化。
使用条件变量进行优化
为了解决上述循环休眠浪费cpu和队列状态发生变化时（已满到未满，已空到未空）被阻塞线程无法及时响应的问题，v3版本引入条件变量对其进行优化。
条件变量由底层的操作系统内核实现的、用于线程间同步的利器。（条件变量的实现原理可以参考我之前的博客：https://www.cnblogs.com/xiaoxiongcanguan/p/14152830.html）
java将不同操作系统内核提供的条件变量机制抽象封装后，作为可重入锁ReentrantLock的附属给程序员使用。且为了避免lost wakeup问题，在条件变量的实现中增加了校验，要求调用条件变量的signal和await方法时当前线程必须先获得条件变量所附属的锁才行，更具体的解析可以参考这篇文章：https://mp.weixin.qq.com/s/ohcr6T1aB7-lVFJIfyJZjA。
引入条件变量后，可以令未满足某种条件的线程暂时进入阻塞态，等待在一个条件变量上；当对应条件满足时由其它的线程将等待在条件变量上的线程唤醒，将其从阻塞态再切换回就绪态。
举个例子：当某一生产者线程想要插入新元素但阻塞队列已满时，可以令当前生产者线程等待并阻塞在对应的条件变量中；当后续某一消费者线程执行出队操作使得队列非空后，将等待在条件变量上的生产者线程唤醒，被唤醒的生产者线程便能及时的再次尝试进行入队操作。
v3和v2版本相比，等待在条件变量进入阻塞态的线程不再周期性的被唤醒而占用过多的cpu资源，且在特定条件满足时也能被及时唤醒。
引入条件变量后的v3版本阻塞队列效率比v2高出许多。
v3版本完整代码：
/**
* 数组作为底层结构的阻塞队列 v3版本
*/
public class MyArrayBlockingQueueV3<E> implements MyBlockingQueue<E> {
/**
* 队列默认的容量大小
* */
private static final int DEFAULT_CAPACITY = 16;
/**
* 承载队列元素的底层数组
* */
private final Object[] elements;
/**
* 当前头部元素的下标
* */
private int head;
/**
* 下一个元素插入时的下标
* */
private int tail;
/**
* 队列中元素个数
* */
private int count;
private final ReentrantLock reentrantLock;
private final Condition condition;
//=================================================构造方法======================================================
/**
* 默认构造方法
* */
public MyArrayBlockingQueueV3() {
this(DEFAULT_CAPACITY);
}
/**
* 默认构造方法
* */
public MyArrayBlockingQueueV3(int initCapacity) {
assert initCapacity > 0;
// 设置数组大小为默认
this.elements = new Object[initCapacity];
// 初始化队列 头部,尾部下标
this.head = 0;
this.tail = 0;
this.reentrantLock = new ReentrantLock();
this.condition = this.reentrantLock.newCondition();
}
/**
* 下标取模
* */
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) throws InterruptedException {
// 先尝试获得互斥锁，以进入临界区
reentrantLock.lockInterruptibly();
try {
// 因为被消费者唤醒后可能会被其它的生产者再度填满队列，需要循环的判断
while (this.count == elements.length) {
// put操作时，如果队列已满则进入条件变量的等待队列，并释放条件变量对应的锁
condition.await();
}
// 走到这里，说明当前队列不满，可以执行入队操作
enqueue(e);
// 唤醒可能等待着的消费者线程
// 由于共用了一个condition，所以不能用signal，否则一旦唤醒的也是生产者线程就会陷入上面的while死循环）
condition.signalAll();
} finally {
// 入队完毕，释放锁
reentrantLock.unlock();
}
}
@Override
public E take() throws InterruptedException {
// 先尝试获得互斥锁，以进入临界区
reentrantLock.lockInterruptibly();
try {
// 因为被生产者唤醒后可能会被其它的消费者消费而使得队列再次为空，需要循环的判断
while(this.count == 0){
condition.await();
}
E headElement = dequeue();
// 唤醒可能等待着的生产者线程
// 由于共用了一个condition，所以不能用signal，否则一旦唤醒的也是消费者线程就会陷入上面的while死循环）
condition.signalAll();
return headElement;
} finally {
// 出队完毕，释放锁
reentrantLock.unlock();
}
}
@Override
public boolean isEmpty() {
return this.count == 0;
}
}
2.4 v4版本（引入双条件变量，优化唤醒效率）
v3版本通过引入条件变量解决了v2版本中循环休眠、唤醒效率低下的问题，但v3版本还是存在一定的性能问题。
v3版本中signalAll的效率问题
jdk的Condition条件变量提供了signal和signalAll这两个方法用于唤醒等待在条件变量中的线程，其中signalAll会唤醒等待在条件变量上的所有线程，而signal则只会唤醒其中一个。
举个例子，v3版本中消费者线程在队列已满时进行出队操作后，通过signalAll会唤醒所有等待入队的多个生产者线程，但最终只会有一个线程成功竞争到互斥锁并成功执行入队操作，其它的生产者线程在被唤醒后发现队列依然是满的，而继续等待。v3版本中的signalAll唤醒操作造成了惊群效应，无意义的唤醒了过多的等待中的线程。
但仅仅将v3版本中的signalAll改成signal是不行的，因为生产者和消费者线程是等待在同一个条件变量中的，如果消费者在出队后通过signal唤醒的不是与之对应的生产者线程，而是另一个消费者线程，则本该被唤醒的生产者线程可能迟迟无法被唤醒，甚至在一些场景下会永远被阻塞，无法再唤醒。
仔细思索后可以发现，对于生产者线程其在队列已满时阻塞等待，等待的是队列不满的条件（notFull）；而对于消费者线程其在队列为空时阻塞等待，等待的是队列不空的条件（notEmpty）。队列不满和队列不空实质上是两个互不相关的条件。
因此v4版本中将生产者线程和消费者线程关注的条件变量拆分成两个：生产者线程在队列已满时阻塞等待在notFull条件变量上，消费者线程出队后通过notFull.signal尝试着唤醒一个等待的生产者线程；与之相对的，消费者线程在队列为空时阻塞等待在notEmpty条件变量上，生产者线程入队后通过notEmpty.signal尝试着唤醒一个等待的消费者线程。
通过拆分出两个互相独立的条件变量，v4版本避免了v3版本中signalAll操作带来的惊群效应，避免了signalAll操作无效唤醒带来的额外开销。
v4版本完整代码：
/**
* 数组作为底层结构的阻塞队列 v4版本
*/
public class MyArrayBlockingQueueV4<E> implements MyBlockingQueue<E> {
/**
* 队列默认的容量大小
* */
private static final int DEFAULT_CAPACITY = 16;
/**
* 承载队列元素的底层数组
* */
private final Object[] elements;
/**
* 当前头部元素的下标
* */
private int head;
/**
* 下一个元素插入时的下标
* */
private int tail;
/**
* 队列中元素个数
* */
private int count;
private final ReentrantLock reentrantLock;
private final Condition notEmpty;
private final Condition notFull;
//=================================================构造方法======================================================
/**
* 默认构造方法
* */
public MyArrayBlockingQueueV4() {
this(DEFAULT_CAPACITY);
}
/**
* 默认构造方法
* */
public MyArrayBlockingQueueV4(int initCapacity) {
assert initCapacity > 0;
// 设置数组大小为默认
this.elements = new Object[initCapacity];
// 初始化队列 头部,尾部下标
this.head = 0;
this.tail = 0;
this.reentrantLock = new ReentrantLock();
this.notEmpty = this.reentrantLock.newCondition();
this.notFull = this.reentrantLock.newCondition();
}
/**
* 下标取模
* */
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) throws InterruptedException {
// 先尝试获得互斥锁，以进入临界区
reentrantLock.lockInterruptibly();
try {
// 因为被消费者唤醒后可能会被其它的生产者再度填满队列，需要循环的判断
while (this.count == elements.length) {
// put操作时，如果队列已满则进入notFull条件变量的等待队列，并释放条件变量对应的互斥锁
notFull.await();
// 消费者进行出队操作时
}
// 走到这里，说明当前队列不满，可以执行入队操作
enqueue(e);
// 唤醒可能等待在notEmpty中的一个消费者线程
notEmpty.signal();
} finally {
// 入队完毕，释放锁
reentrantLock.unlock();
}
}
@Override
public E take() throws InterruptedException {
// 先尝试获得互斥锁，以进入临界区
reentrantLock.lockInterruptibly();
try {
// 因为被生产者唤醒后可能会被其它的消费者消费而使得队列再次为空，需要循环的判断
while(this.count == 0){
notEmpty.await();
}
E headElement = dequeue();
// 唤醒可能等待在notFull中的一个生产者线程
notFull.signal();
return headElement;
} finally {
// 出队完毕，释放锁
reentrantLock.unlock();
}
}
@Override
public boolean isEmpty() {
return this.count == 0;
}
}
2.5 v5版本（引入双锁令生产者和消费者能并发操作阻塞队列）
v4版本的阻塞队列采用双条件变量之后，其性能已经不错了，但仍存在进一步优化的空间。
v4版本单锁的性能问题
v4版本中阻塞队列的出队、入队操作是使用同一个互斥锁进行并发同步的，这意味着生产者线程和消费者线程无法并发工作，消费者线程必须等待生产者线程操作完成退出临界区之后才能继续执行，反之亦然。单锁的设计在生产者和消费者都很活跃的高并发场景下会一定程度限制阻塞队列的吞吐量。
因此v5版本在v4版本的基础上，将出队和入队操作使用两把锁分别管理，使得生产者线程和消费者线程可以并发的操作阻塞队列，达到进一步提高吞吐量的目的。
使用两把锁分别控制出队、入队后，还需要一些调整来解决生产者/消费者并发操作队列所带来的问题。
存在并发问题的双锁版本出队、入队实现第一版（v4基础上的微调）：
/**
this.takeLock = new ReentrantLock();
this.notEmpty = this.takeLock.newCondition();
this.putLock = new ReentrantLock();
this.notFull = this.putLock.newCondition();
*/
@Override
public void put(E e) throws InterruptedException {
// 先尝试获得互斥锁，以进入临界区
putLock.lockInterruptibly();
try {
// 因为被消费者唤醒后可能会被其它的生产者再度填满队列，需要循环的判断
while (this.count == elements.length) {
// put操作时，如果队列已满则进入notFull条件变量的等待队列，并释放条件变量对应的互斥锁
notFull.await();
}
// 走到这里，说明当前队列不满，可以执行入队操作
enqueue(e);
// 唤醒可能等待在notEmpty中的一个消费者线程
notEmpty.signal();
} finally {
// 入队完毕，释放锁
putLock.unlock();
}
}
@Override
public E take() throws InterruptedException {
// 先尝试获得互斥锁，以进入临界区
takeLock.lockInterruptibly();
try {
// 因为被生产者唤醒后可能会被其它的消费者消费而使得队列再次为空，需要循环的判断
while(this.count == 0){
notEmpty.await();
}
E headElement = dequeue();
// 唤醒可能等待在notFull中的一个生产者线程
notFull.signal();
return headElement;
} finally {
// 出队完毕，释放锁
takeLock.unlock();
}
}
上面基于v4版本微调的双锁实现虽然容易理解，但由于允许消费者和生产者线程并发的访问队列而存在几个严重问题。
1. count属性线程不安全
队列长度count字段是一个用于判断队列是否已满，队列是否为空的重要属性。在v5之前的版本count属性一直被唯一的同步锁保护着，任意时刻至多只有一个线程可以进入临界区修改count的值。而引入双锁令消费者线程/生产者线程能并发访问后，count变量的自增/自减操作会出现线程不安全的问题。
解决方案：将int类型的count修改为AtomicInteger来解决生产者/消费者同时访问、修改count时导致的并发问题。
2. 生产者/消费者线程死锁问题
在上述的代码示例中，生产者线程首先获得生产者锁去执行入队操作，然后唤醒可能阻塞在notEmpty上的消费者线程。由于使用条件变量前首先需要获得其所属的互斥锁，如果生产者线程不先释放生产者锁就去获取消费者的互斥锁，那么就存在出现死锁的风险。消费者线程和生产者线程可以并发的先分别获得消费者锁和生产者锁，并且也同时尝试着获取另一把锁，这样双方都在等待着对方释放锁，互相阻塞出现死锁现象。
解决方案：先释放已获得的锁之后再去获得另一个锁执行唤醒操作
存在并发问题的双锁版本出队、入队实现第二版（在上述第一版基础上进行微调）：
/**
private final AtomicInteger count = new AtomicInteger();
this.takeLock = new ReentrantLock();
this.notEmpty = this.takeLock.newCondition();
this.putLock = new ReentrantLock();
this.notFull = this.putLock.newCondition();
*/
@Override
public void put(E e) throws InterruptedException {
int currentCount;
// 先尝试获得互斥锁，以进入临界区
putLock.lockInterruptibly();
try {
// 因为被消费者唤醒后可能会被其它的生产者再度填满队列，需要循环的判断
while (count.get() == elements.length) {
// put操作时，如果队列已满则进入notFull条件变量的等待队列，并释放条件变量对应的互斥锁
notFull.await();
// 消费者进行出队操作时
}
// 走到这里，说明当前队列不满，可以执行入队操作
enqueue(e);
currentCount = count.getAndIncrement();
} finally {
// 入队完毕，释放锁
putLock.unlock();
}
// 如果插入之前队列为空，才唤醒等待弹出元素的线程
if (currentCount == 0) {
signalNotEmpty();
}
}
@Override
public E take() throws InterruptedException {
E headElement;
int currentCount;
// 先尝试获得互斥锁，以进入临界区
takeLock.lockInterruptibly();
try {
// 因为被生产者唤醒后可能会被其它的消费者消费而使得队列再次为空，需要循环的判断
while(this.count.get() == 0){
notEmpty.await();
}
headElement = dequeue();
currentCount = this.count.getAndDecrement();
} finally {
// 出队完毕，释放锁
takeLock.unlock();
}
// 只有在弹出之前队列已满的情况下才唤醒等待插入元素的线程
if (currentCount == elements.length) {
signalNotFull();
}
return headElement;
}
/**
* 唤醒等待队列非空条件的线程
*/
private void signalNotEmpty() {
// 为了唤醒等待队列非空条件的线程，需要先获取对应的takeLock
takeLock.lock();
try {
// 唤醒一个等待非空条件的线程
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
/**
* 唤醒等待队列未满条件的线程
*/
private void signalNotFull() {
// 为了唤醒等待队列未满条件的线程，需要先获取对应的putLock
putLock.lock();
try {
// 唤醒一个等待队列未满条件的线程
notFull.signal();
} finally {
putLock.unlock();
}
}
3. lost wakeup问题
在上述待改进的双锁实现第二版中，阻塞在notFull中的生产者线程完全依赖相对应的消费者线程来将其唤醒（阻塞在notEmpty中的消费者线程也同样依赖对应的生产者线程将其唤醒），这在生产者线程和消费者线程并发时会出现lost wakeup的问题。
下面构造一个简单而不失一般性的例子来说明，为什么上述第二版的实现中会出现问题。
时序图（假设阻塞队列的长度为5（element.length=5），且一开始时队列已满）
| 生产者线程P1 | 生产者线程P2 | 消费者线程C | |
| 1 | 执行put操作，此时队列已满。 执行while循环中的notfull.await陷入阻塞状态 （await会释放putLock） | ||
| 2 | 执行take操作，队列未满，成功执行完dequeue。 此时currentCount=5，this.count=4， 执行takeLock.unLock释放takeLock锁 | ||
| 3 | 执行put操作，拿到putLock锁，由于消费者C已经执行完出队操作， 成功执行enqueue。 此时currentCount=4，this.count=5, 执行putLock.unLock释放putLock锁 | ||
| 4 | 判断currentCount == elements.length为真， 执行signalNotFull，并成功拿到putLock。 notFull.signal唤醒等待在其上的生产者线程P1。 take方法执行完毕，return返回 | ||
| 5 | 被消费者C唤醒，但此时count=5，无法跳出while循环， 继续await阻塞在notFull条件变量中 | ||
| 6 | 判断currentCount == 0为假，进行处理。 put方法执行完毕 ，return返回 |
可以看到，虽然生产者线程P1由于队列已满而先被阻塞，而消费者线程C在出队后也确实通知唤醒了生产者线程P1。但是由于生产者线程P2和消费者线程C的并发执行，导致了生产者线程P1在被唤醒后依然无法成功执行入队操作，只能继续的阻塞下去。在一些情况下，P1生产者线程可能再也不会被唤醒而永久的阻塞在条件变量notFull上。
为了解决这一问题，双锁版本的阻塞队列其生产者线程不能仅仅依靠消费者线程来将其唤醒，而是需要在其它生产者线程在入队操作完成后，发现队列未满时也尝试着唤醒由于上述并发场景发生lost wakeup问题的生产者线程(消费者线程在出队时的优化亦是如此)。
最终优化的V5版本的出队、入队实现：
/**
private final AtomicInteger count = new AtomicInteger();
this.takeLock = new ReentrantLock();
this.notEmpty = this.takeLock.newCondition();
this.putLock = new ReentrantLock();
this.notFull = this.putLock.newCondition();
*/
@Override
public void put(E e) throws InterruptedException {
int currentCount;
// 先尝试获得互斥锁，以进入临界区
putLock.lockInterruptibly();
try {
// 因为被消费者唤醒后可能会被其它的生产者再度填满队列，需要循环的判断
while (count.get() == elements.length) {
// put操作时，如果队列已满则进入notFull条件变量的等待队列，并释放条件变量对应的互斥锁
notFull.await();
// 消费者进行出队操作时
}
// 走到这里，说明当前队列不满，可以执行入队操作
enqueue(e);
currentCount = count.getAndIncrement();
// 如果在插入后队列仍然没满，则唤醒其他等待插入的线程
if (currentCount + 1 < elements.length) {
notFull.signal();
}
} finally {
// 入队完毕，释放锁
putLock.unlock();
}
// 如果插入之前队列为空，才唤醒等待弹出元素的线程
// 为了防止死锁，不能在释放putLock之前获取takeLock
if (currentCount == 0) {
signalNotEmpty();
}
}
@Override
public E take() throws InterruptedException {
E headElement;
int currentCount;
// 先尝试获得互斥锁，以进入临界区
takeLock.lockInterruptibly();
try {
// 因为被生产者唤醒后可能会被其它的消费者消费而使得队列再次为空，需要循环的判断
while(this.count.get() == 0){
notEmpty.await();
}
headElement = dequeue();
currentCount = this.count.getAndDecrement();
// 如果队列在弹出一个元素后仍然非空，则唤醒其他等待队列非空的线程
if (currentCount - 1 > 0) {
notEmpty.signal();
}
} finally {
// 出队完毕，释放锁
takeLock.unlock();
}
// 只有在弹出之前队列已满的情况下才唤醒等待插入元素的线程
// 为了防止死锁，不能在释放takeLock之前获取putLock
if (currentCount == elements.length) {
signalNotFull();
}
return headElement;
}
/**
* 唤醒等待队列非空条件的线程
*/
private void signalNotEmpty() {
// 为了唤醒等待队列非空条件的线程，需要先获取对应的takeLock
takeLock.lock();
try {
// 唤醒一个等待非空条件的线程
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
/**
* 唤醒等待队列未满条件的线程
*/
private void signalNotFull() {
// 为了唤醒等待队列未满条件的线程，需要先获取对应的putLock
putLock.lock();
try {
// 唤醒一个等待队列未满条件的线程
notFull.signal();
} finally {
putLock.unlock();
}
}
3. 不同版本阻塞队列的性能测试
前面从v2版本开始，对所实现的阻塞队列进行了一系列的优化，一直到最终的V5版本实现了一个基于双锁，双条件变量的高性能版本。
下面对v3-v5版本进行一轮基础的性能测试（v2无限轮询性能太差），看看其实际性能是否真的如博客第二章中所说的那般，高版本的性能是更优秀的。同时令jdk中的ArrayBlockingQueue和LinkedBlockingQueue也实现MyBlockingQueue，也加入测试。
测试工具类BlockingQueueTestUtil：
public class BlockingQueueTestUtil {
public static long statisticBlockingQueueRuntime(
MyBlockingQueue<Integer> blockingQueue, int workerNum, int perWorkerProcessNum, int repeatTime) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(workerNum * 2);
// 第一次执行时存在一定的初始化开销，不进行统计
oneTurnExecute(executorService,blockingQueue,workerNum,perWorkerProcessNum);
long totalTime = 0;
for(int i=0; i<repeatTime; i++){
long oneTurnTime = oneTurnExecute(executorService,blockingQueue,workerNum,perWorkerProcessNum);
totalTime += oneTurnTime;
}
executorService.shutdown();
assert blockingQueue.isEmpty();
return totalTime/repeatTime;
}
private static long oneTurnExecute(ExecutorService executorService, MyBlockingQueue<Integer> blockingQueue,
int workerNum, int perWorkerProcessNum) throws InterruptedException {
long startTime = System.currentTimeMillis();
CountDownLatch countDownLatch = new CountDownLatch(workerNum * 2);
// 创建workerNum个生产者/消费者
for(int i=0; i<workerNum; i++){
executorService.execute(()->{
produce(blockingQueue,perWorkerProcessNum);
countDownLatch.countDown();
});
executorService.execute(()->{
consume(blockingQueue,perWorkerProcessNum);
countDownLatch.countDown();
});
}
countDownLatch.await();
long endTime = System.currentTimeMillis();
return endTime - startTime;
}
private static void produce(MyBlockingQueue<Integer> blockingQueue,int perWorkerProcessNum){
try {
// 每个生产者生产perWorkerProcessNum个元素
for(int j=0; j<perWorkerProcessNum; j++){
blockingQueue.put(j);
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
private static void consume(MyBlockingQueue<Integer> blockingQueue,int perWorkerProcessNum){
try {
// 每个消费者消费perWorkerProcessNum个元素
for(int j=0; j<perWorkerProcessNum; j++){
blockingQueue.take();
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
jdk的ArrayBlockingQueue简单包装（JDKArrayBlockingQueue）：
public class JDKArrayBlockingQueue<E> implements MyBlockingQueue<E> {
private final BlockingQueue<E> jdkBlockingQueue;
/**
* 指定队列大小的构造器
*
* @param capacity 队列大小
*/
public JDKArrayBlockingQueue(int capacity) {
if (capacity <= 0)
throw new IllegalArgumentException();
jdkBlockingQueue = new ArrayBlockingQueue<>(capacity);
}
@Override
public void put(E e) throws InterruptedException {
jdkBlockingQueue.put(e);
}
@Override
public E take() throws InterruptedException {
return jdkBlockingQueue.take();
}
@Override
public boolean isEmpty() {
return jdkBlockingQueue.isEmpty();
}
@Override
public String toString() {
return "JDKArrayBlockingQueue{" +
"jdkBlockingQueue=" + jdkBlockingQueue +
'}';
}
}
jdk的LinkedBlockingQueue简单包装（JDKLinkedBlockingQueue）：
public class JDKLinkedBlockingQueue<E> implements MyBlockingQueue<E> {
private final BlockingQueue<E> jdkBlockingQueue;
/**
* 指定队列大小的构造器
*
* @param capacity 队列大小
*/
public JDKLinkedBlockingQueue(int capacity) {
if (capacity <= 0)
throw new IllegalArgumentException();
jdkBlockingQueue = new LinkedBlockingQueue<>(capacity);
}
@Override
public void put(E e) throws InterruptedException {
jdkBlockingQueue.put(e);
}
@Override
public E take() throws InterruptedException {
return jdkBlockingQueue.take();
}
@Override
public boolean isEmpty() {
return jdkBlockingQueue.isEmpty();
}
@Override
public String toString() {
return "JDKLinkedBlockingQueue{" +
"jdkBlockingQueue=" + jdkBlockingQueue +
'}';
}
}
测试主体代码：
public class BlockingQueuePerformanceTest {
/**
* 队列容量
* */
private static final int QUEUE_CAPACITY = 3;
/**
* 并发线程数（消费者 + 生产者 = 2 * WORKER_NUM）
* */
private static final int WORKER_NUM = 30;
/**
* 单次测试中每个线程访问队列的次数
* */
private static final int PER_WORKER_PROCESS_NUM = 3000;
/**
* 重复执行的次数
* */
private static final int REPEAT_TIME = 5;
public static void main(String[] args) throws InterruptedException {
{
MyBlockingQueue<Integer> myArrayBlockingQueueV3 = new MyArrayBlockingQueueV3<>(QUEUE_CAPACITY);
long avgCostTimeV3 = BlockingQueueTestUtil.statisticBlockingQueueRuntime(myArrayBlockingQueueV3, WORKER_NUM, PER_WORKER_PROCESS_NUM, REPEAT_TIME);
System.out.println(costTimeLog(MyArrayBlockingQueueV3.class, avgCostTimeV3));
}
{
MyBlockingQueue<Integer> myArrayBlockingQueueV4 = new MyArrayBlockingQueueV4<>(QUEUE_CAPACITY);
long avgCostTimeV4 = BlockingQueueTestUtil.statisticBlockingQueueRuntime(myArrayBlockingQueueV4, WORKER_NUM, PER_WORKER_PROCESS_NUM, REPEAT_TIME);
System.out.println(costTimeLog(MyArrayBlockingQueueV4.class, avgCostTimeV4));
}
{
MyBlockingQueue<Integer> myArrayBlockingQueueV5 = new MyArrayBlockingQueueV5<>(QUEUE_CAPACITY);
long avgCostTimeV5 = BlockingQueueTestUtil.statisticBlockingQueueRuntime(myArrayBlockingQueueV5, WORKER_NUM, PER_WORKER_PROCESS_NUM, REPEAT_TIME);
System.out.println(costTimeLog(MyArrayBlockingQueueV5.class, avgCostTimeV5));
}
{
MyBlockingQueue<Integer> jdkArrayBlockingQueue = new JDKArrayBlockingQueue<>(QUEUE_CAPACITY);
long avgCostTimeJDK = BlockingQueueTestUtil.statisticBlockingQueueRuntime(jdkArrayBlockingQueue, WORKER_NUM, PER_WORKER_PROCESS_NUM, REPEAT_TIME);
System.out.println(costTimeLog(JDKArrayBlockingQueue.class, avgCostTimeJDK));
}
{
MyBlockingQueue<Integer> jdkLinkedBlockingQueue = new JDKLinkedBlockingQueue<>(QUEUE_CAPACITY);
long avgCostTimeJDK = BlockingQueueTestUtil.statisticBlockingQueueRuntime(jdkLinkedBlockingQueue, WORKER_NUM, PER_WORKER_PROCESS_NUM, REPEAT_TIME);
System.out.println(costTimeLog(JDKLinkedBlockingQueue.class, avgCostTimeJDK));
}
}
private static String costTimeLog(Class blockQueueCLass,long costTime){
return blockQueueCLass.getSimpleName() + " avgCostTime=" + costTime + "ms";
}
}
上述代码指定的参数为基于最大容量为3的阻塞队列，生产者、消费者线程各30个，每个线程执行3000次出队或入队操作，重复执行5次用于统计平均时间。
我的机器上的运行结果如下:
MyArrayBlockingQueueV3 avgCostTime=843ms
MyArrayBlockingQueueV4 avgCostTime=530ms
MyArrayBlockingQueueV5 avgCostTime=165ms
JDKArrayBlockingQueue avgCostTime=506ms
JDKLinkedBlockingQueue avgCostTime=163ms执行时长v3 > v4 > JDKArrayBlockingQueue > MyArrayBlockingQueueV5 > JDKLinkedBlockingQueue，且v4耗时大致等于JDKArrayBlockingQueue、v5耗时大致等于JDKLinkedBlockingQueue。
究其原因是因为jdk的ArrayBlockingQueue实现和V4版本一样，是基于单锁，双条件变量的；而jdk的LinkedBlockingQueue实现和V5版本一样，是基于双锁，双条件变量的（V4、V5版本的实现就是参考的jdk源码）。
虽然测试的用例不是很全面，但测试结果和理论大致是吻合的，希望大家通过测试结果来加深对不同版本间性能差异的背后原理的理解。
4. 为什么jdk中的ArrayBlockingQueue不基于性能更好的双锁实现 ？
看到这里，不知你是否和我一样对为什么jdk的ArrayBlockingQueue使用单锁而不使用性能更好的双锁实现而感到疑惑。所幸网上也有不少小伙伴有类似的疑问，这里将相关内容简单梳理一下。
1. 基于数组实现的阻塞队列（ABQ）是可以采用双锁实现更加高效率的出队、入队的。但由于jdk中阻塞队列是属于集合Collection的一个子类，双锁版本的ABQ其迭代器会比单锁的复杂很多很多，但在性能上的改善并不那么的可观。ABQ的实现在复杂度和性能上做了一个折中，选择了容易实现但性能稍低的单锁实现。
http://jsr166-concurrency.10961.n7.nabble.com/ArrayBlockingQueue-concurrent-put-and-take-tc1306.html
2. 如果对性能有更加苛刻要求的话，可以考虑使用jdk中基于双锁实现的LinkedBlockingQueue（LBQ）。需要注意的是，在高吞吐量的出队、入队的场景下，LBQ链式的结构在垃圾回收时性能会略低于基于数组的，紧凑结构的ABQ。
3. jdk提供了一个庞大而全面的集合框架，每个具体的数据结构都需要尽可能多的实现高层的接口和抽象方法。这样的设计对于使用者来说确实很友好，但也令实现者背上了沉重的负担，必须为实现一些可能极少使用的接口而花费巨大的精力，甚至反过来影响到特定数据结构的本身的实现。ABQ受制于双锁版本迭代器实现的复杂度，而被迫改为效率更低的单锁实现就是一个典型的例子。
5. 总结
前段时间迷上了MIT6.824的数据库课程，在理解了课程所提供的实验后（共6个lab）收获很大，因此想着自己再动手实现一个更加全面的版本（并发的B+树，MVCC多版本控制、行级锁以及sql解释器、网络协议等等）。但一段时间后发现上述的功能难度很大且实现起来细节很多，这将耗费我过多的时间而被迫放弃了（膨胀了Orz）。在被打击后，清醒的意识到对于现阶段的我来说还是应该稳扎稳打，着眼于更小的知识点，通过自己动手造轮子的方式加深对知识点的理解，至于撸一个完善的关系型数据库这种宏大的目标受制于我目前的水平还是暂时先放放吧。
本篇博客的完整代码在我的github上：https://github.com/1399852153/Reinventing-the-wheel-for-learning（blocking queue模块）。后续应该会陆续更新关于自己动手实现线程池、抽象同步队列AQS等的博客。
还存在很多不足之处，请多多指教。
锋哥最新SpringCloud分布式电商秒杀课程发布
👇👇👇
👆长按上方微信二维码 2 秒
感谢点赞支持下哈 