LinkedBlockingQueue实现原理
LinkedBlockingQueue 是用链表实现的有界阻塞队列,当构造对象时为指定队列大小时,队列默认大小为Integer.MAX_VALUE
。从它的构造方法可以看出:
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
LinkedBlockingQueue的主要属性
LinkedBlockingQueue 的主要属性有:
/** Current number of elements */
private final AtomicInteger count = new AtomicInteger();
/**
* Head of linked list.
* Invariant: head.item == null
*/
transient Node<E> head;
/**
* Tail of linked list.
* Invariant: last.next == null
*/
private transient Node<E> last;
/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();
可以看出与 ArrayBlockingQueue 主要的区别是,LinkedBlockingQueue 在插入数据和删除数据时分别是由两个不同的lock(takeLock
和putLock
)来控制线程安全的,因此,也由这两个lock生成了两个对应的condition(notEmpty
和notFull
)来实现可阻塞的插入和删除数据。并且,采用了链表的数据结构来实现队列,Node 结点的定义为:
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; }
}
put方法详解
put方法源码为:
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
/*
* Note that count is used in wait guard even though it is
* not protected by lock. This works because count can
* only decrease at this point (all other puts are shut
* out by lock), and we (or some other waiting put) are
* signalled if it ever changes from capacity. Similarly
* for all other uses of count in other wait guards.
*/
//如果队列已满,则阻塞当前线程,将其移入等待队列
while (count.get() == capacity) {
notFull.await();
}
//入队操作,插入数据
enqueue(node);
c = count.getAndIncrement();
//若队列满足插入数据的条件,则通知被阻塞的生产者线程
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
put 方法的逻辑也同样很容易理解,可见注释。基本上和 ArrayBlockingQueue 的 put 方法一样。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;
}
ArrayBlockingQueue与LinkedBlockingQueue的比较
相同点:ArrayBlockingQueue和LinkedBlockingQueue都是通过condition通知机制来实现可阻塞式插入和删除元素,并满足线程安全的特性;
不同点:1. ArrayBlockingQueue底层是采用的数组进行实现,而 LinkedBlockingQueue 则是采用链表数据结构; 2. ArrayBlockingQueue插入和删除数据,只采用了一个lock,而 LinkedBlockingQueue 则是在插入和删除分别采用了putLock
和takeLock
,这样可以降低线程由于线程无法获取到lock而进入 WAITING 状态的可能性,从而提高了线程并发执行的效率。