阻塞队列 LinkedBlockingQueue
介绍
上篇介绍了应用CAS算法实现的非阻塞队列ConcurrentLinkedQueue，本篇介绍的是应用独占锁实现的阻塞队列LinkedBlockingQueue。
该类图能够看到 LinkedBlockingQueue 也是应用单向链表实现的，其中蕴含head Node，last Node，用来寄存头尾节点；并且还有一个初始值为 0 的原子变量count，用来记录队列元素个数；另外还蕴含两个 ReentrantLock 实例，别离用来管制元素入队和出队的原子性，其中 takeLock 用来管制同一时刻只有一个线程能够从队列头部获取元素，putLock 就是管制同一时刻只有一个线程能够从队列尾部增加元素。
notEmpty 和 notFull 是条件变量，他们外部都有一个条件队列来寄存进队和出队时被阻塞的线程。
LinkedBlockingQueue 的构造函数如下：
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);}能够看到默认状况下，LinkedBlockingQueue 队列容量为 int 最大值，当然咱们也能够间接指定容量大小，所以在肯定水平上能够阐明 LinkedBlockingQueue 是有界阻塞队列。
offer 操作
向队列尾部插入一个元素，如果队列有闲暇则插入胜利返回 true，如果队列已满则抛弃以后元素返回 false。如果插入的元素为 null 则抛出异样。并且该办法是非阻塞的。
public boolean offer(E e) { //（1）元素为空则抛出异样 if (e == null) throw new NullPointerException(); //（2）如果队列已满则抛弃元素 final AtomicInteger count = this.count; if (count.get() == capacity) return false; int c = -1; //（3）构建新节点，而后获取putLock独占锁 Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; putLock.lock(); try { //（4）如果队列不满则进入队列，并减少计数 if (count.get() < capacity) { enqueue(node); c = count.getAndIncrement(); //（5） if (c + 1 < capacity) notFull.signal(); } } finally { //（6）开释锁 putLock.unlock(); } //（7） if (c == 0) signalNotEmpty(); //（8） return c >= 0;}代码（1）首先判断入队元素是否空，为空抛出异样。
代码（2）则判断是否队列已满，满的话就抛弃并返回 false。
代码（3）构建一个新的 Node 节点，而后获取 putLock 锁，当获取锁后其余线程调用 put 或 offer 办法都会被阻塞，放入 putLock 的 AQS 阻塞队列中。
代码（4）判断队列是否已满，为什么要再判断一次呢？因为在执行代码（2）到获取锁之间可能有其它线程通过 put 或 offer 增加了新元素，所以再判断一次是否队列已满，不满则新元素入队并且减少计数器。
代码（5）判断如果新元素入队后还有残余空间，则唤醒 notFull 条件队列中正在阻塞的一个线程，（唤醒因为调用 notFull 的 await 操作的线程，比方执行了 put 办法而队列已满的时候）。
代码（6）开释获取的 putLock 锁，这里要留神，锁的开释肯定要在 finally 外面做，因为即便 try 块抛出异样了，finally 也是会被执行到。另外开释锁后其余因为调用 put 操作而被阻塞的线程将会有一个获取到该锁。
代码（7）c == 0阐明在执行代码（6）开释锁时队列至多有一个元素，队列外面有元素则执行 signalNotEmpty 办法。
private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); }}该办法的作用就是激活 notEmpty 的条件队列中因为调用 notEmpty 的 await 办法（比方调用 take 办法并且队列为空的时候）而被阻塞的一个线程，这也阐明了调用条件变量的办法前要获取对应的锁。
总结：offer 办法通过 putLock 锁来保障增加的原子性，另外须要留神的就是在调用条件变量的时候须要先获取到对应的锁，并且入队只操作链表的尾结点。
put 操作
该办法向队列尾部插入一个元素，如果队列闲暇则插入胜利后间接返回，如果队列已满则阻塞以后线程，直到队列有闲暇后插入胜利返回。如果在阻塞时被其余线程设置了中断标记，则被阻塞线程会抛出异样而后返回。如果传入的元素是 null 则抛出异样。
public void put(E e) throws InterruptedException { //（1）插入元素为空抛出异样 if (e == null) throw new NullPointerException(); int c = -1; //（2）构建新节点，并获取独占锁putLock Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { //（3）如果队列满则期待 while (count.get() == capacity) { notFull.await(); } //（4）插入队列并递增计数 enqueue(node); c = count.getAndIncrement(); //（5） if (c + 1 < capacity) notFull.signal(); } finally { //（6） putLock.unlock(); } //（7） if (c == 0) signalNotEmpty();}咱们能够看到在代码（2）中应用putLock.lockInterruptibly()获取独占锁，相比在 offer 办法中获取独占锁的办法这个办法能够被中断。具体地说就是以后线程在获取锁的过程中，如果被其余线程设置了中断标记则以后线程会抛出 InterruptedException 异样，所以 put 操作在获取锁的过程中是可被中断的。
代码（3）判断以后队列已满，则调用notFull.await()办法把以后线程放入 notFull 的条件队列中，而后以后队列会开释 putLock 锁，因为 putLock 锁被开释了，所以别的线程就有机会获取到该锁。
应用 while 而不是 if 是为了避免虚伪唤醒问题
poll 操作
从队列头部获取并移除一个元素，如果队列为空则返回 null，该办法是非阻塞办法。
public E poll() { final AtomicInteger count = this.count; //（1）队列为空则返回null if (count.get() == 0) return null; //（2）获取独占锁 E x = null; int c = -1; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { //（3）队列不为空则出队并且递加计数 if (count.get() > 0) { x = dequeue(); c = count.getAndDecrement(); //（4） if (c > 1) notEmpty.signal(); } } finally { //（5） takeLock.unlock(); } //（6） if (c == capacity) signalNotFull(); //（7） return x;}private E dequeue() { Node<E> h = head; Node<E> first = h.next; h.next = h; // help GC head = first; E x = first.item; first.item = null; return x;}首先代码（1）很简略，先判断一下以后队列是否为空，为空间接返回 null。
代码（2）则获取独占锁 takeLock，以后线程获取该锁后，其余线程在调用 poll 或 take 办法时会被阻塞。
代码（3）判断如果以后队列不为空则进行出队操作，而后递加计数器。
尽管代码（3）中的判断队列是否为空和获取队列元素不是原子性的，但只有在 poll、take 或者 remove 操作的中央会递加 count 计数值，然而这三个办法都须要获取到 takeLock 锁能力进行操作，而以后线程曾经获取了 takeLock 锁，所以其余线程没有机会在当前情况下递加 count 计数值，所以看起来不是原子性的，然而它们是线程平安的。
代码（4）判断如果c > 1则阐明以后线程移除掉队列外面的一个元素后队列不为空（c 是删除元素前队列元素个数），那么这时候就能够激活因为调用 take 办法而被阻塞到 notEmpty 的条件队列外面的一个线程。
代码（6）阐明以后线程移除队头元素前以后队列是满的，移除队头元素后以后队列至多有一个闲暇地位，那么这时候就能够调用 signalNotFull 激活因为调用 put 办法而被阻塞到 notFull 的条件队列里的一个线程，signalNotFull 的代码如下。
private void signalNotFull() { final ReentrantLock putLock = this.putLock; putLock.lock(); try { notFull.signal(); } finally { putLock.unlock(); }}peek 操作
获取队列头部元素然而不从队列外面移除它，如果队列为空则返回 null。该办法是非阻塞办法。
public E peek() { //(1) if (count.get() == 0) return null; //(2) final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { Node<E> first = head.next; //(3) if (first == null) return null; else //(4) return first.item; } finally { takeLock.unlock(); }}peek 办法不是很简单，须要留神的就是这里须要再判断一下first == null，不能间接返回 first.item，因为代码（1）和代码（2）不是原子性的，很可能在代码（1）判断队列不为空后在获取锁前有其它线程执行了 poll 或 take 操作导致队列为空，而后间接返回 fist.item 就会空指针异样。
take 操作
该办法获取以后队列头部元素并从队列中移除它，如果队列为空则阻塞以后线程直到队列不为空而后获取并返回元素，如果在阻塞时被其余线程设置了中断标记，则阻塞线程会抛出异样。
public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; //（1）获取锁 takeLock.lockInterruptibly(); try { //（2）队列为空则阻塞挂起 while (count.get() == 0) { notEmpty.await(); } //（3）出队并递加计数 x = dequeue(); c = count.getAndDecrement(); //（4） if (c > 1) notEmpty.signal(); } finally { //（5） takeLock.unlock(); } //（6） if (c == capacity) signalNotFull(); return x;}在代码（1）中，以后线程获取到独占锁，其余调用 take 或者 poll 操作的线程将会被阻塞挂起。
代码（2）判断如果队列为空则阻塞挂起以后线程，并把以后线程放入 notEmpty 的条件队列中。
代码（3）进行出队操作并递加计数。
代码（4）判断如果c > 1则阐明以后队列不为空，那么唤醒 notEmpty 的条件队列外面的一个因为调用 take 操作而被阻塞的线程。
代码（5）开释锁。
代码（6）判断如果c == capacity则阐明以后队列至多有一个闲暇地位，那就激活 notFull 的条件队列外面的一个因为调用 put 操作而被阻塞的线程。
remove 操作
删除队列外面指定的元素，有则删除并返回 true，没有则返回 false。
public boolean remove(Object o) { if (o == null) return false; //（1）获取putLock 和 takeLock fullyLock(); try { //（2）遍历寻找要删除的元素 for (Node<E> trail = head, p = trail.next; p != null; trail = p, p = p.next) { //（3） if (o.equals(p.item)) { unlink(p, trail); return true; } } //（4） return false; } finally { //（5） fullyUnlock(); }}void fullyLock() { putLock.lock(); takeLock.lock();}首先代码（1）获取到双重锁，而后其余线程的入队和出队操作都会被挂起。
代码（2）遍历队列寻找要删除的元素，找不到返回 false，找到执行 unlink 办法，咱们看一下这个办法。
void unlink(Node<E> p, Node<E> trail) { p.item = null; trail.next = p.next; if (last == p) last = trail; //如果以后队列满，再删除后也要唤醒期待的线程 if (count.getAndDecrement() == capacity) notFull.signal();}trail 为删除元素的前驱节点，删除元素后，如果发现以后队列有闲暇空间，则唤醒 notFull 的条件队列中的一个因为调用 put 办法而被阻塞的线程。
代码（5）调用 fullyUnlock 来进行开释双重锁。
void fullyUnlock() { takeLock.unlock(); putLock.unlock();}总结：因为 remove 办法在删除指定元素前加了两把锁，所以在遍历队列查找指定元素的过程中是线程平安的，并且此时其余调用入队、出队操作的线程全副会被阻塞。留神，获取多个资源锁的程序与开释的程序是相同的。
size 操作
获取以后队列元素个数。
public int size() { return count.get();}因为出队和入队操作的 count 都是加锁的，所以后果相比 ConcurrentLinkedQueue 的 size 办法来说是精确的。
总结
LinkedBlockingQueue 的外部是通过单向链表实现的，应用头、尾节点来进行入队和出队操作，也就是入队操作都是对尾节点进行操作，出队操作都是对头节点进行操作。
对头、尾节点的操作别离应用了独自的独占锁从而保障了原子性，所以出队和入队操作是能够同时进行的。另外对头、尾节点的独占锁都装备了一个条件队列，用来寄存被阻塞的线程，并联合入队、出队操作实现了一个生产生产模型。