LinkedBlockingQueue源码分析（JDK8）

1. 基于链表实现，线程安全的阻塞队列。
2. “two lock queue”算法变体，双锁（ReentrantLock）：takeLock、putLock，允许读写并行，remove(e)和迭代器iterators需要获取2个锁。
3. FIFO先进先出模式。
4. 在大部分并发场景下，LinkedBlockingQueue的吞吐量比ArrayBlockingQueue更好。

static class Node<E> {E item; // 数据Node<E> next; // 下一个节点的指针，默认为null意味为最后一个节点，无后继Node(E x) { item = x; }
}

// 将node链接到队列尾部
private void enqueue(Node<E> node) { // 入队// assert putLock.isHeldByCurrentThread();// assert last.next == null;last = last.next = node; // 等价于last.next = node;last = last.next(即node)
}
public LinkedBlockingQueue(Collection<? extends E> c) {this(Integer.MAX_VALUE);final ReentrantLock putLock = this.putLock;putLock.lock(); // Never contended(竞争), but necessary for visibility(可见性)try {int n = 0;for (E e : c) {if (e == null)throw new NullPointerException();if (n == capacity)throw new IllegalStateException("Queue full");enqueue(new Node<E>(e)); // 执行last = last.next = node;++n;}count.set(n); // 设置队列元素个数} finally {putLock.unlock();}
}

/**
* 唤醒notEmpty上正在等待获取元素的线程
*/
private void signalNotEmpty() {final ReentrantLock takeLock = this.takeLock;takeLock.lock();try {notEmpty.signal();} finally {takeLock.unlock();}
}

// 队列已满:false
public boolean offer(E e) {if (e == null) throw new NullPointerException();final AtomicInteger count = this.count;if (count.get() == capacity) // 队列容量达到最大值，添加失败return false;int c = -1;Node<E> node = new Node<E>(e);final ReentrantLock putLock = this.putLock;putLock.lock(); // 获取插入锁putLocktry {if (count.get() < capacity) { // 加锁后再次判断队列是否已满enqueue(node); // 入队c = count.getAndIncrement(); // 返回Inc之前的值if (c + 1 < capacity) // 插入节点后队列未满notFull.signal(); // 唤醒notFull上的等待线程}} finally {putLock.unlock(); // 释放插入锁}if (c == 0)signalNotEmpty(); // 如果offer前队列为空，则唤醒notEmpty上的等待线程return c >= 0;
}

public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {}

long nanos = unit.toNanos(timeout);  //将指定的时间长度转换为毫秒来进行处理
while (count.get() == capacity) {if (nanos <= 0) // 等待的剩余时间小于等于0，那么直接返回falsereturn false;nanos = notFull.awaitNanos(nanos); // 最多等待时间（纳秒）
}

public boolean add(E e) {if (offer(e))return true;elsethrow new IllegalStateException("Queue full");
}

// 一直阻塞直到插入成功
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;// 可中断的锁获取操作（优先考虑响应中断），如果线程由于获取锁而处于Blocked状态时，线程将被中断而不再继续等待（throws InterruptedException），可避免死锁。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循环可避免“伪唤醒”（线程被唤醒时队列大小依旧达到最大值）while (count.get() == capacity) {notFull.await(); // notFull：入队条件}enqueue(node); // 将node链接到队列尾部c = count.getAndIncrement(); // 元素入队后队列元素总和if (c + 1 < capacity) // 队列未满notFull.signal(); // 唤醒其他执行入队列的线程} finally {putLock.unlock(); // 释放锁}
// c=0说明队列之前为空，出队列线程均处于等待状态。添加一个元素后，队列已不为空，于是唤醒等待获取元素的线程if (c == 0)signalNotEmpty();
}

/*** 唤醒等待插入数据的线程. Called only from take/poll.*/
private void signalNotFull() {final ReentrantLock putLock = this.putLock;putLock.lock();try {notFull.signal();} finally {putLock.unlock();}
}
/**
* 队列头部元素出队.
*
* @return the node
*/
private E dequeue() {// assert takeLock.isHeldByCurrentThread();// assert head.item == null;Node<E> h = head; // 临时变量hNode<E> first = h.next;h.next = h; // 形成环引用help GChead = first;E x = first.item;first.item = null;return x;
}

// 队列为空返回null而不是抛异常
public E poll() {final AtomicInteger count = this.count;if (count.get() == 0)return null;E x = null;int c = -1;final ReentrantLock takeLock = this.takeLock;takeLock.lock();try {if (count.get() > 0) {x = dequeue();c = count.getAndDecrement(); // 减1并返回旧值if (c > 1)notEmpty.signal(); // 唤醒其他取数据的线程}} finally {takeLock.unlock();}// c等于capacity说明poll之前队列已满，poll一个元素后便可唤醒其他等待插入数据的线程if (c == capacity)signalNotFull();return x;
}

public E poll(long timeout, TimeUnit unit)throws InterruptedException{}

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) // take数据前队列大小大于1，则take后队列至少还有1个元素notEmpty.signal(); // 唤醒其他取数据的线程} finally {takeLock.unlock();}if (c == capacity)signalNotFull(); //唤醒其他等待插入数据的线程return x;
}

public int drainTo(Collection<? super E> c, int maxElements) {if (c == null)throw new NullPointerException();if (c == this)throw new IllegalArgumentException();if (maxElements <= 0)return 0;boolean signalNotFull = false;final ReentrantLock takeLock = this.takeLock;takeLock.lock();try {int n = Math.min(maxElements, count.get());//转移元素数量不能超过队列总量 // count.get provides visibility to first n NodesNode<E> h = head;int i = 0;try {while (i < n) {Node<E> p = h.next;//从队首获取元素c.add(p.item);p.item = null;//p为临时变量，置null方便GCh.next = h;h = p;++i;}return n;} finally {// Restore invariants even if c.add() threwif (i > 0) { // 有数据被转移到集合c中// assert h.item == null;head = h;//如果转移前的队列大小等于队列容量，则说明现在队列未满// 更新count为队列实际大小（减去i得到）signalNotFull = (count.getAndAdd(-i) == capacity);}}} finally {takeLock.unlock();if (signalNotFull)signalNotFull(); // 唤醒其他等待插入数据的线程}
}

 public int drainTo(Collection<? super E> c) {return drainTo(c, Integer.MAX_VALUE);
}

public boolean retainAll(Collection<?> c) {Objects.requireNonNull(c); // 集合为null则throw NPEboolean modified = false;Iterator<E> it = iterator();while (it.hasNext()) {if (!c.contains(it.next())) {it.remove();modified = true; // 队列因此请求而改变则返回true}}return modified;
}

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)return null;elsereturn first.item;} finally {takeLock.unlock();}
}

/*** Locks to prevent both puts and takes.*/
void fullyLock() {putLock.lock();takeLock.lock();
}
/*** Unlocks to allow both puts and takes.*/
void fullyUnlock() {takeLock.unlock();putLock.unlock();
}