说明

通过阅读源码,了解LinkedBlockingQueue的特性。本文基于JDK1.7源码

正文

通过查询API对LinkedBlockingQueue特点进行简单的了解:

  • LinkedBlockingQueue是一个基于已链接节点的,范围任意的blocking queue
  • 此队列按FIFO(先进先出)排序元素
  • 新元素插入到队列的尾部,并且队列获取操作会获得位于队列头部的元素
  • 链接队列的吞吐量通常要高于基于数组的对列(ArrayBlockingQueue),但是在大多数并发应用程序中,其可预知的性能要低
  • 可选的容量范围构造方法参数作为防止队列过度扩展的一种方法,如果未指定容量,则等于Integer.MAX_VALUE,除非插入节点会使队列超出容量,否则每次插入后会动态地创建链接节点

LinkedBlockingQueue

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {

LinkedBlockingQueue继承自AbstractQueue,实现了BlockingQueue,Serializable接口
关于BlockingQueue接口在《深入理解ArrayBlockingQueue》一文中已有介绍,在此不再赘述

节点内部类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; }
}

LinkedBlockingQueue的属性

    private final int capacity; // 队列容量,如果构造时未指定则为Integer.MAX_VALUE

    //使用AtomicInteger来统计队列中元素数量
private final AtomicInteger count = new AtomicInteger(0); /**
* Head of linked list.
* Invariant: head.item == null
*/
private transient Node<E> head; // 队列的头元素 值为null /**
* Tail of linked list.
* Invariant: last.next == null
*/
private transient Node<E> last; // 队列的尾元素 它的下一个节点为null /** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock(); // 出队锁 /** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition(); // 取出线程condition /** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock(); // 入队锁 /** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition(); // 添加线程condition

LinkedBlockingQueue的构造函数

public LinkedBlockingQueue() {
this(Integer.MAX_VALUE); //当未指定容量时,为Integer.MAX_VALUE
}
  • 1
  • 2
  • 3
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null); // 构造队列时创建值为null的节点
}

LinkedBlockingQueue的添加方法

put(e)

该方法没有返回值,当队列已满时,会阻塞当前线程

 public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();//先检查添加值是否为null
// 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); //先创建新的节点
final ReentrantLock putLock = this.putLock; //使用putLock来保证线程安全
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try { while (count.get() == capacity) {//当队列已满,添加线程阻塞
notFull.await();
}
enqueue(node); // 调用enqueue方法添加到队尾
c = count.getAndIncrement(); //调用AtomicInteger的getAndIncrement()是数量加1
if (c + 1 < capacity)//添加成功后判断是否可以继续添加,队列未满
notFull.signal(); //唤醒添加线程
} finally {
putLock.unlock();
}
if (c == 0) // 添加后如果队列中只有一个元素,唤醒一个取出线程,使用取出锁
signalNotEmpty();
}

offer(e,timeout,unit)

该方法返回true或false,当队列已满时,会阻塞给定时间,添加操作成功返回true,否则返回false

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; //使用putLock
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) { //当队列已满阻塞给定时间
if (nanos <= 0) //当时间消耗完全,操作未成功 返回false
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e)); // 调用enqueue方法添加一个新的节点
c = count.getAndIncrement(); //同样调用AtomicInteger的方法
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true; // 操作成功返回true
}

offer(e)

该方法返回true或false,不会阻塞,直接返回

 public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false; //当队列已满,直接返回false
int c = -1;
Node<E> node = new Node(e); // 先创建新的节点
final ReentrantLock putLock = this.putLock;//使用putLock
putLock.lock();
try {
if (count.get() < capacity) { // 加锁后再次判断队列是否已满
enqueue(node); //调用enqueue方法将节点添加到队尾
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0; // 比较c的大小,判断是否成功,当c大于-1时则添加操作成功
}

通过以上三种添加方法我们发现:

  • 添加时,使用putLock这个可重入锁保证线程安全
  • 共同调用了enqueue()方法实现节点的添加
  • 改变队列元素数量时,调用了AtomicInteger的getAndIncrement()方法,保证原子性

enqueue(e)

 /**
* Links node at end of queue.
*
* @param node the node
*/
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
last = last.next = node; //将新的节点添加到队尾,并变成新的尾节点
}

getAndIncrement()

此方法内部调用了compareAndSet()方法

public final int getAndIncrement() {
for (;;) { // 采用循环的方式,将值加1
int current = get();
int next = current + 1;
if (compareAndSet(current, next))
return current; //加1成功后返回原值
}
}

compareAndSet( expect, update)
此方法内部调用了unsafe类的compareAndSwapInt()方法,该方法共有四个参数,第一个参数为需要修改的对象,第二个为偏移量(内存的旧值),第三个参数为期待的值,第四个为更新后的值;若果valueOffset的值和expect的值相等,则将valueOffset值修改为update返回true,否则不做操作,返回false

public final boolean compareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

LinkedBlockingQueue的取出方法

take()

 public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;//使用takeLock保证线程安全
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {//当队列为空,取出线程阻塞
notEmpty.await();
}
x = dequeue(); //掉用dequeue方法从队头取出元素
c = count.getAndDecrement(); //调用AtomicInteger的getAndDecrement()将count值减1
if (c > 1)//判断如果当前队列之前元素的数量大于1,唤醒取出线程
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)//之前队列元素数量为容量值,取出一个,只能唤醒一个添加线程
signalNotFull();
return x;
}

poll(timeout,unit)

该方法取出元素时,如果队列为空,则阻塞给定的时间

  public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;//使用takeLock保证线程安全
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {//当队列为空则阻塞给定时间
if (nanos <= 0)//时间消耗完全后,如果操作未成功则返回null
return null;
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();//调用dequeue方法返回节点值
c = count.getAndDecrement();//将count值减1
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}

poll()

该方法取出元素时,如果队列为空,则直接返回null

 public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)// 如果队列为空,直接返回null
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;//使用takeLock保证线程安全
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();//调用dequeue方法取出队头节点元素的值
c = count.getAndDecrement();//count减1
if (c > 1)//如果取出元素不是唯一的,唤醒取出线程
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
if (c == capacity)//如果从已满队列取出的,则唤醒一个添加线程
signalNotFull();
return x;
}

通过以上三种取出方法,发现:

  • 所有的取出方法都是使用takeLock这个可重入锁保证线程安全
  • 都调用了dequeue()方法从队头取出元素
  • 调用了AtomicInteger的getAndDecrement()方法将count值减1,保证原子性

dequeue()

/**
* Removes a node from head of queue.
*
* @return the node
*/
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head; //队列的头结点是值为null的节点
Node<E> first = h.next; //返回头节点之后的第一个节点
h.next = h; // help GC
//因为创建节点时创建了一个新的对象,所以需要GC,即需要将头节点的后继节点指向自身,帮助GC
head = first;//将新的头节点置为将要删除的第一个节点
E x = first.item; //将节点的值赋给x
first.item = null;//将节点值置为null,变为新的头节点
return x;//返回取出的值
}

getAndDecrement()

此方法与getAndIncrement()方法相似,底层都是调用了unsafe的compareAndSwapInt方法保证操作的原子性

peek()

该方法只返回队头元素的值,并不能将节点从队列中删除

public E peek() {
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null)//如果队列为空,则直接返回null
return null;
else
return first.item;
} finally {
takeLock.unlock();
}
}

remove(o)

从队列中删除指定元素值的节点

public boolean remove(Object o) {
if (o == null) return false;
fullyLock(); //此时将入队锁和出队锁全部锁住来保证线程安全
try {
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {// 循环遍历查找值相等的元素
if (o.equals(p.item)) {
unlink(p, trail);//调用unlink删除此节点
return true;//操作成功返回true
}
}
return false;
} finally {
fullyUnlock();
}
}

fullyLock()

void fullyLock() {
putLock.lock();
takeLock.lock();
}

unlink(p, trail)

 void unlink(Node<E> p, Node<E> trail) {//p为要删除节点,trail为删除节点的前一个节点
// assert isFullyLocked();
// p.next is not changed, to allow iterators that are
// traversing p to maintain their weak-consistency guarantee.
p.item = null;
trail.next = p.next; // 改变指针将前一节点的后继节点指向删除节点的后一个节点
if (last == p)
last = trail;
if (count.getAndDecrement() == capacity)
notFull.signal();
}

获取队列当前大小及剩余容量

size()

 public int size() {
return count.get();
}

remainingCapacity()

public int remainingCapacity() {
return capacity - count.get();
}

这两个方法都没有使用锁来保证线程安全,是因为count自身为AtomicInteger对象,保证了操作的原子性

05-11 22:38