LinkedBlockingQueue源码分析

LinkedBlockingQueue是基于链表的阻塞队列，其类定义如下所示

static class Node<E> {//节点的定义
E item;

Node<E> next;

Node(E x) { item = x; }
}

/** 队列的容量,默认为Integer.MAX_VALUE */
private final int capacity;

/** 队列中元素的个数 */
private final AtomicInteger count = new AtomicInteger(0);

/**
* 队列的头指针，take（）操作修改头指针，并由锁takeLock 保护
* Invariant: head.item == null
*/
private transient Node<E> head;

/**
* 队列的尾指针，put()操作修改尾指针，由锁putLock 来保护
* 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();

在LinkedBlockingQueue类的实现中，很重要的一个和ArrayBlockingQueue不同的地方，是对put()和take()分别使用了两个不同的锁，put()使用了putLock来同步，修改尾指针last。take()使用takeLock来同步，修改头指针head。而针对“空”和“满”的阻塞条件，也是对这两个所对象分别构建的两个Condition对象（notEmpty和notFull），构成了双锁双条件。

对于put()和take()以及类似的操作，双锁避免了互相影响，一定意义上看，减小了操作的锁粒度，提高了并发性。

put()操作如下：

public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {

while (count.get() == capacity) {
notFull.await();//队列满时，阻塞于notFull条件变量
}
enqueue(node);//添加元素，修改last指针
c = count.getAndIncrement();
//比较迷惑的是下面的if语句，该语句用来唤醒阻塞的put线程，下面做详细介绍
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}

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中put线程在每次put(）完后，都会调用 notEmpty.signal()，唤醒阻塞于notEmpty上的take()线程。而在LinkedBlockingQueue中唤醒操作主要发生在同类型线程之间，put线程唤醒put线程，take线程唤醒take线程。

例如：队列容量为5，当前队列满：

【1】->【2】->【3】->【4】->【5】

此时来了3个put线程，put1,put2,put3.因为队列满，所以三个put线程全部阻塞于notFull上。然后来了3个take线程，take1,take2,take3。当take1线程执行到take()操作中的if (c == capacity) signalNotFull();时唤醒一个阻塞的put线程，假设put1线程被唤醒。此时count=4。接着take2线程执行take()操作，由于count=4 , take2线程不会唤醒任何put线程。接着take3线程执行take()操作，由于count=3, take3线程不会唤醒任何put线程。此时count=2，接下来put1线程执行put()操作，c的值为2，执行if (c + 1 < capacity) notFull.signal();唤醒阻塞的put2线程。put2线程执行put()操作，c的值为3,执行if (c + 1 < capacity) notFull.signal();唤醒阻塞的put3线程.

由以上例子可以看出，唤醒操作主要发生在同类型操作之间。

注意：调用条件变量的await，signal时，前提是线程必须获取到于条件变量相关联的锁。