LinkedBlockingQueue是基于单向链表实现的有界阻塞队列，队列元素遵循FIFO，LinkedBlockingQueue比基于数组的阻塞队列拥有更好的吞吐量，但是在大部分并发应用中，性能不如基于数组的队列。
和ArrayBlocingQueue一样继承AbstractQueue，实现BlockingQueue接口，不再看BlockingQueue接口代码，直接看LinkedBlockingQueue的结构吧。
//单向链表的节点
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; }
}
/** The capacity bound, or Integer.MAX_VALUE if none */
private final int capacity; //链表的容量，如果构造不传入，那就最大
/** Current number of elements */
private final AtomicInteger count = new AtomicInteger(0); //元素的个数
private transient Node<E> head;
private transient Node<E> last;
//因为阻塞队列，从head取元素，从tail存元素，链表采用2个lock分别对应存取
//2个对应的条件队列对应非空和非满
/** 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();
//构造，如果不传入容量大小，默认最大，还是传入的好
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null); //初始head和last指向一个元素为null的Node
}
看下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); //新建一个node
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly(); //响应中断的lock
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) { //put时，元素数量等于容量大小，那就条件队列等待
notFull.await();
}
enqueue(node); //将node加入链表
c = count.getAndIncrement();
if (c + 1 < capacity) //容量未到达限制，可以继续加入，通知notfull条件队列
notFull.signal();
} finally {
putLock.unlock(); //释放锁
}
//这里c默认是-1，经过enqueue入队，c=count.getAndIncrement(),如果变成0，说明链表队列原来是空的，现在有元素了
//所以这里可以唤醒一个notEmpty条件队列种线程，take元素
if (c == 0)
signalNotEmpty();
}
//将元素加入链表
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
last = last.next = node; //只需要修改last的next，然后将last指向新加入的node
}
看下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(); //为0，队列空，那就加入notEmpyt条件队列
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1) //c大于1说明还有元素，那就可以继续take，所以可以唤醒
notEmpty.signal();
} finally {
takeLock.unlock();
}
//默认c为-1，经过take后c=count.getAndDecrement();说明原来队列是满的，take后不满，就可以唤醒notFull条件队列
if (c == capacity)
signalNotFull();
return x;
}
//返回的是head指向的next的元素，并且断开与原head的链接，head指向原head的next
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head; //初始化的时候head的item为null
Node<E> first = h.next; //第一个真正有值的元素
h.next = h; // help GC ，head的next指向自己
head = first;
E x = first.item;
first.item = null;
return x;
}
/**
* Signals a waiting take. Called only from put/offer (which do not
* otherwise ordinarily lock takeLock.)
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
/**
* Signals a waiting put. Called only from take/poll.
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
其他的offer，poll都比较简单，大都能看懂。不过这些获取存储都是使用单个的锁，LinkedBlockingQueue里面有个remove操作使用的是2把锁：
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);
return true;
}
}
return false;
} finally {
fullyUnlock();
}
}
void fullyLock() {
putLock.lock();
takeLock.lock();
}
/**
* Unlock to allow both puts and takes.
*/
void fullyUnlock() {
takeLock.unlock();
putLock.unlock();
}
void unlink(Node<E> p, Node<E> 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();
}
因为remove操作需要遍历整个链表，所以加2把锁遍历。