java并发编程之深入学习Concurrent包(十二，阻塞队列.1)
yuyutoo 2024-12-12 15:54 2 浏览 0 评论
引言：
java.util.concurrent.BlockingQueue阻塞队列，通常用于一个线程生产对象，并放入队列，另外一个线程获取并消费这些对象的场景，很多消息框架都有类似实现。
接口实现方法简介：
如下图所示：
入队时，因容器限制导致插入异常，使用add会抛出IllegalStateException，使用offer则直接返回false，使用put则会线程阻塞直到可以入队。
出队情况一样。
接口实现类简介：
BlockingQueue是个接口，目前有以下几个类实现了BlockingQueue接口：
ArrayBlockingQueue：
一个内部元素由数组存放的有界阻塞队列。初始化的时候必须设定容量，用先进先出(FIFO)的方式对元素进行入队和出队。
DelayQueue：
DelayQueue中内部使用的是PriorityQueue存放数据，其中的元素必须实现Delayed接口，可以判断元素的延迟是否达到，如果达到则该元素将会在DelayQueue的下一次take被调用的时候出队。
LinkedBlokingQueue：
无界阻塞队列，以链表结构对元素进行存储，用先进先出(FIFO)的方式对元素进行入队和出队。
PriorityBlockingQueue：
有优先级的无界阻塞队列，元素必须实现java.lang.Comparable接口，用来作为优先级的比较。因此在该队列中元素的排序取决于元素类中的实现。
SynchronousQueue：
同步队列，这是一个特殊的队列，内部一次只能容纳一个元素。如果该队列已存在一个元素，往该队列中插入一个元素的线程将会被阻塞，直到其他线程将该队列内的元素从队列中出队。
ArrayBlockingQueue源码实现：
1.属性值
final Object[] items;
int takeIndex;
int putIndex;
int count;
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
如上所示，数据存放在items这个数组，入队和出队的索性各一个，当前容量count，入队出队共享的锁lock，非空的条件Condition及非满的条件Condition。
2.入队操作(以offer为例)
步骤一，检查元素e不为null，获取超时时间，获取可重入锁。
步骤二，如果count=items.length(队列的数组已经存满)：如果超时时间为负，则失败；否则线程在。
步骤三，进入enqueue方法，添加元素，增加count值，唤醒在notEmpty这个Condition的等待队列上等待读取队列数据的线程。
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
checkNotNull(e);
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
3.出队操作(poll为例)
步骤一，获取超时时间，获取可重入锁。
步骤二，如果count == 0(容量为空)。如果超时时间为负，则失败；notEmpty这个Condition的等待队列中 的情况，延迟时间到依然需要等待while循环结束
步骤三，进入dequeue方法，删除元素，count值，唤醒在notEmpty这个Condition的等待队列上等待将数据入队的线程。
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
return dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return x;
}
总结：
如下图所示，ArrayBlockingQueue在一个给定的数组中，通过两个index进行循环，做到入队出队的操作。
队列为空时，使用notEmpty这个Condition让队列出队的线程阻塞等待。
当队列满时，使用notFull这个Condition让队列入队的线程阻塞等待。
入队出队使用的是同一个lock，同一时间只能做一个，因为是重入锁，持有锁的同个线程再次进入可直接进入。
Condition操作可参考本人之前结束的文章。
LinkedBlokingQueue源码实现：
1.属性介绍
static class Node<E> {
E item;
/**
* One of:
* - the real successor Node
* - this Node, meaning the successor is head.next
* - null, meaning there is no successor (this is the last node)
*/
Node<E> next;
Node(E x) { item = x; }
}
private final int capacity;
private final AtomicInteger count = new AtomicInteger();
transient Node<E> head;
private transient Node<E> last;
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
如上代码所示，由Node节点构成的链表结构
head和last作为头尾，因为是transient类型，可以判断头尾不是存放具体元素的节点。
capacity是容量，为空时，容量默认是Integer.MAX_VALUE。
锁有takeLock和putLock两个，入队和出队各自处理，不相互影响。
count容量因可能有入队出队两个线程操作，使用原子操作类。
2.入队操作(以offer为例)
从如下代码可以看出，与ArrayBlockingQueue的入队操作基本类似，只有以下几点差异：
1.使用putLock防止入队操作争用，对出队操作无影响。
2.链表插入处理与数组不同
3.count原子操作类，累加方法特殊。
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}
private void enqueue(Node<E> node) {
last = last.next = node;
}
出队操作差不多，不再赘述。
PriorityBlockingQueue源码实现：
1.基本属性实现
private static final int DEFAULT_INITIAL_CAPACITY = 11;
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
private transient Object[] queue;
private transient int size;
private transient Comparator<? super E> comparator;
private final ReentrantLock lock;
private final Condition notEmpty;
private transient volatile int allocationSpinLock;
private PriorityQueue<E> q;
如上代码所示：
默认队列容量11
最大的队列容量为Integer.MAX_VALUE - 8
queue是队列元素存放的数组，size是当前队列中元素的个数
comparator是元素比较器，比较结果将决定元素的队列顺序
lock是出队入队的共享锁
allocationSpinLock是扩容数组分配资源时的自旋锁，CAS需要使用。
2.入队过程(以offer为例)
如下代码所示，
在入队时，判断容量是否满足需要，如果不满足，使用tryGrow扩容。
siftUpComparable是真正的插入操作方法。
而在siftUpComparable中，逻辑上使用二叉树的方式进行数据存储和获取，从而实现有优先级的获取功能。
siftUpComparable过程如下所示：
先插入到最后节点，再和父节点比较大小，小于则内容互换(结构不变，内容替换)。