在 Java8 中，提供了 7 个阻塞队列
- ArrayBlockingQueue 数组实现的有界阻塞队列, 此队列按照先进先出（FIFO）的原则
对元素进行排序。 - LinkedBlockingQueue 链表实现的有界阻塞队列, 此队列的默认和最大长度为
Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序 - PriorityBlockingQueue 支持优先级排序的无界阻塞队列, 默认情况下元素采取自然顺序升序排列。也可以自定义类实现 compareTo()方法来指定元素排序规则，或者初始化 PriorityBlockingQueue 时，指定构造参数 Comparator 来对元素进行排序。
- DelayQueue 优先级队列实现的无界阻塞队列
- SynchronousQueue 不存储元素的阻塞队列, 每一个 put 操作必须等待一个 take 操作，否则不能继续添加元素。
- LinkedTransferQueue 链表实现的无界阻塞队列
- LinkedBlockingDeque 链表实现的双向阻塞队列
在阻塞队列中，提供了四种处理方式
-
插入操作
add(e) ：添加元素到队列中，如果队列满了，继续插入元素会报错，IllegalStateException。
offer(e) : 添加元素到队列，同时会返回元素是否插入成功的状态，如果成功则返回 true
put(e) ：当阻塞队列满了以后，生产者继续通过 put添加元素，队列会一直阻塞生产者线程，知道队列可用
offer(e,time,unit) ：当阻塞队列满了以后继续添加元素，生产者线程会被阻塞指定时间，如果超时，则线程直接退出 -
移除操作
remove()：当队列为空时，调用 remove 会返回 false，如果元素移除成功，则返回 true
poll(): 当队列中存在元素，则从队列中取出一个元素，如果队列为空，则直接返回 null
take()：基于阻塞的方式获取队列中的元素，如果队列为空，则 take 方法会一直阻塞，直到队列中有新的数据可以消费
poll(time,unit)：带超时机制的获取数据，如果队列为空，则会等待指定的时间再去获取元素返回
ArrayBlockingQueue 源码分析
构造方法
ArrayBlockingQueue 提供了三个构造方法，分别如下：
capacity： 表示数组的长度，也就是队列的长度
fair：表示是否为公平的阻塞队列，默认情况下构造的是非公平的阻塞队列。
其中第三个构造方法就不解释了，它提供了接收一个几个作为数据初始化的方法
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
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(); //初始化非满等待队列
}
关于锁的用途，大家在没有看接下来的源码之前，可以先思考一下他的作用。
items 构造以后，大概是一个这样的数组结构
image.png
Add 方法
以 add 方法作为入口，在 add 方法中会调用父类的 add 方法，也就是 AbstractQueue.如果看源码看得比较多的话，一般这种写法都是调用父类的模版方法来解决通用性问题
public boolean add(E e) {
return super.add(e);
}
从父类的 add 方法可以看到，这里做了一个队列是否满了的判断，如果队列满了直接抛出一个异常
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
offer 方法
add 方法最终还是调用 offer 方法来添加数据，返回一个添加成功或者失败的布尔值反馈
这段代码做了几个事情
- 判断添加的数据是否为空
- 添加重入锁
- 判断队列长度，如果队列长度等于数组长度，表示满了直接返回 false
- 否则，直接调用 enqueue 将元素添加到队列中
public boolean offer(E e) {
checkNotNull(e); //对请求数据做判断
final ReentrantLock lock = this.lock;
lock.lock();
try {
f (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}
enqueue
这个是最核心的逻辑，方法内部通过 putIndex 索引直接将元素添加到数组 items
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x; //通过 putIndex 对数据赋值
if (++putIndex == items.length) // 当putIndex 等于数组长度时，将 putIndex 重置为 0
putIndex = 0;
count++;//记录队列元素的个数
notEmpty.signal();//唤醒处于等待状态下的线程，表示当前队列中的元素不为空,如果存在消费者线程阻塞，就可以开始取出元素
}
putIndex 为什么会在等于数组长度的时候重新设置为 0?
因为 ArrayBlockingQueue 是一个 FIFO 的队列，队列添加元素时，是从队尾获取 putIndex 来存储元素，当 putIndex等于数组长度时，下次就需要从数组头部开始添加了
下面这个图模拟了添加到不同长度的元素时，putIndex 的变化，当 putIndex 等于数组长度时，不可能让 putIndex 继续累加，否则会超出数组初始化的容量大小。同时大家还需要思考两个问题:
- 当元素满了以后是无法继续添加的，因为会报错
- 其次，队列中的元素肯定会有一个消费者线程通过 take或者其他方法来获取数据，而获取数据的同时元素也会从队列中移除
image.png
put 方法
put 方法和 add 方法功能一样，差异是 put 方法如果队列满了，会阻塞。
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); //这个也是获得锁，但是和 lock 的区别是，这个方法优先允许在等待时由其他线程调用等待线程的 interrupt 方法来中断等待直接返回。而 lock方法是尝试获得锁成功后才响应中断
try {
while (count == items.length)
notFull.await();//队列满了的情况下，当前线程将会被 notFull 条件对象挂起加到等待队列中
enqueue(e);
} finally {
lock.unlock();
}
}
image.png
take 方法
take 方法是一种阻塞获取队列中元素的方法
它的实现原理很简单，有就删除没有就阻塞，注意这个阻塞是可以中断的，如果队列没有数据那么就加入 notEmpty条件队列等待(有数据就直接取走，方法结束)，如果有新的put 线程添加了数据，那么 put 操作将会唤醒 take 线程，执行 take 操作。
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await(); //如果队列为空的情况下，直接通过 await 方法阻塞
return dequeue();
} finally {
lock.unlock();
}
}
image.png
如果队列中添加了元素，那么这个时候，会在 enqueue 中调用 notempty.signal 唤醒 take 线程来获得元素
image.png
dequeue 方法
这个是出队列的方法，主要是删除队列头部的元素并发返回给客户端
takeIndex，是用来记录拿数据的索引值
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex]; //默认获取 0 位置的元素
items[takeIndex] = null;//将该位置的元素设置为空
if (++takeIndex == items.length)//这里的作用也是一样，如果拿到数组的最大值，那么重置为 0，继续从头部位置开始获取数据
takeIndex = 0;
count--;//记录 元素个数递减
if (itrs != null)
itrs.elementDequeued();//同时更新迭代器中的元素数据
notFull.signal();//触发 因为队列满了以后导致的被阻塞的线程
return x;
}
itrs.elementDequeued()
ArrayBlockingQueue 中，实现了迭代器的功能，也就是可以通过迭代器来遍历阻塞队列中的元素
所以 itrs.elementDequeued() 是用来更新迭代器中的元素数据的
takeIndex 的索引变化图如下，同时随着数据的移除，会唤醒处于 put 阻塞状态下的线程来继续添加数据
image.png
remove 方法
remove 方法是移除一个指定元素
public boolean remove(Object o) {
if (o == null) return false;
final Object[] items = this.items; //获取数组元素
final ReentrantLock lock = this.lock;
lock.lock(); //获得锁
try {
if (count > 0) { //如果队列不为空
final int putIndex = this.putIndex;
//获取下一个要添加元素时的索引
int i = takeIndex;//获取当前要被移除的元素的索引
do {
if (o.equals(items[i])) {//从takeIndex 下标开始，找到要被删除的元素
removeAt(i);//移除指定元素
return true;//返回执行结果
}
//当前删除索引执行加 1 后判断是否与数组长度相等
//若为 true，说明索引已到数组尽头，将 i 设置为 0
if (++i == items.length)
i = 0;
} while (i != putIndex);//继续查找，直到找到最后一个元素
}
return false;
} finally {
lock.unlock();
}
}
——学自咕泡学院
网友评论