1 前言
这节我们就来看看LinkedBlockingQueue内部实现的原理。
2 LinkedBlockingQueue的使用
在看原理之前我们先来用一用LinkedBlockingQueue，来体验一下：
2.1 插入数据
public class LinkedBlockingQueueTest { public static void main(String[] args) throws InterruptedException { testPut(); } public static void testPut() throws InterruptedException { // 创建一个容量是3的阻塞队列LinkedBlockingQueue BlockingQueue<String> blockingDeque = new LinkedBlockingQueue<String>(3); // 尝试插入10条数据 for (int i = 0; i < 10; i++) { // 往阻塞队列里面插入一条数据 blockingDeque.put(i+""); // 每次成功插入一条数据，打印插入第几条 System.out.println("插入第" + i +"个元素"); } // 插入完成，打印日志 System.out.println("插入元素完成"); } }
上面创建一个容量为3的LinkedBlockingQueue，在插入3条数据之后，阻塞队列满了。这个时候调用put方法插入数据的时候，线程被阻塞住了，所以导致后面的插入被阻塞住了。
2.2 获取数据
public class LinkedBlockingQueueTest { public static void main(String[] args) throws InterruptedException { testTake(); } public static void testTake() throws InterruptedException { // 创建一个容量是3的阻塞队列LinkedBlockingQueue BlockingQueue<String> blockingDeque = new LinkedBlockingQueue<String>(3); blockingDeque.put("A"); blockingDeque.put("B"); blockingDeque.put("C"); for (int i = 0; i < 10; i++) { String e = blockingDeque.take(); System.out.println("获取到阻塞队列第" + i +"个元素为: " + e); } System.out.println("获取元素完成"); } }
事先往阻塞队列里面插入了3个元素，然后调用for循环从阻塞队列里面取10个元素。当3个元素取完之后，阻塞队列为空，这个时候调用take方法获取元素的线程被阻塞了。
2.3 多线程使用LinkedBlockingQueue
public class LinkedBlockingQueueTest { static class PutThread extends Thread { BlockingQueue<String> blockingQueue; public PutThread(BlockingQueue<String> blockingQueue) { this.blockingQueue = blockingQueue; } @Override public void run() { try { testPut(blockingQueue); } catch (InterruptedException e) { e.printStackTrace(); } } } static class TakeThread extends Thread { BlockingQueue<String> blockingQueue; public TakeThread(BlockingQueue<String> blockingQueue) { this.blockingQueue = blockingQueue; } @Override public void run() { try { testTake(blockingQueue); } catch (InterruptedException e) { e.printStackTrace(); } } } public static void testTake(BlockingQueue<String> blockingDeque) throws InterruptedException { for (int i = 0; i < 10; i++) { String e = blockingDeque.take(); System.out.println("获取到阻塞队列第" + i +"个元素为: " + e); } System.out.println("获取元素完成"); } public static void testPut(BlockingQueue<String> blockingDeque) throws InterruptedException { for (int i = 0; i < 10; i++) { blockingDeque.put(i+""); System.out.println("插入第" + i +"个元素"); } System.out.println("插入元素完成"); } public static void main(String[] args) throws InterruptedException { // 创建一个容量是3的阻塞队列LinkedBlockingQueue BlockingQueue<String> blockingDeque = new LinkedBlockingQueue<String>(3); // 创建一个插入数据线程 PutThread putThread = new PutThread(blockingDeque); // 创建一个取数据线程 TakeThread takeThread = new TakeThread(blockingDeque); // 启动插入数据线程 putThread.start(); // 沉睡一秒 Thread.sleep(1000); // 启动获取数据线程 takeThread.start(); } }
上边执行的大致过程：
（1）当putThread线程启动之后，往队列中插入3个元素，此时队列满了，插入操作被阻塞
（2）当takeThread线程启动之后，不断的从队列中取元素出来，此时队列容量又不满了，将插入线程唤醒，此时putThread再次可以插入元素
（3）最后putThread和takeThread完成了插入10条数据，获取10条数据的操作
针对上边的执行原理我们大致画个图理解一下：
上图棕色部分表示插入的步骤；蓝色部分表示获取数据步骤
（1）首先LinkedBlockingQueue有两个锁，插入锁和拉取锁；插入元素之前要获取插入锁，拉取元素之前要获取拉取锁
（2）线程A执行插入操作的时候先获取到插入锁，获取锁成功之后才能进行下一步操作，判断队列的容量是否满了，如果满了则要进入插入等待队列，进行阻塞等待
（3）同样的道理，线程B拉取元素的时候，先要获取拉取锁，获取锁成功之后判断队列是不是空的，如果是空说明没有可获取的元素，需要进行拉取等待队列阻塞等待
（4）当线程A执行插入操作的时候，发现队列容量未满，这个时候可以插入元素，插入完元素之后，此时队列肯定是非空的，所以此时要唤醒一下拉取等待队列中的线程，叫他们醒来，此时队列不是空了，可以获取数据了。
（5）同样的道理，当线程B拉取元素线程发现队列非空，这个时候拉取一个元素出来，此时队列刚刚扣减一个元素，肯定不是满的，所以此时需要唤醒一下插入等待队列中的线程，叫它们起来，队列有位置可以插入了，赶紧起来插入数据。
接下来我们就来通过源码具体看看实现的过程。
3 LinkedBlockingQueue内部源码
3.1 内部结构
先来看下LinkedBlockingQueue的内部结构：
public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { private static final long serialVersionUID = -6903933977591709194L; // 阻塞队列内部是使用链表存储数据的 // 这个Node就是链表的节点 static class Node<E> { // 节点存储的值 E item; // 指向的下一个节点 Node<E> next; // 构造函数 Node(E x) { item = x; } } // 阻塞队列的容量，默认不传是Integer.MAX_INTEGER private final int capacity; // 阻塞队列内部的元素个数，也就是当前队列的大小 private final AtomicInteger count = new AtomicInteger(); // 阻塞队列的头节点 transient Node<E> head; // 阻塞队列的尾节点 private transient Node<E> last; // 阻塞队列内部有两把锁，分别为putLock和takeLock，就是你上面图说的那两把锁 // putLock：将数据放入阻塞队列的时候需要加putLock，这里就是插入锁 private final ReentrantLock takeLock = new ReentrantLock(); // Condition等待条件：notEmpty表示非空条件 // 当阻塞队列为空时，notEmpty非空 条件不满足 // 这个时候从阻塞队列取数据的线程就会被阻塞等待 private final Condition notEmpty = takeLock.newCondition(); // 阻塞队列内部有两把锁，分别为putLock和takeLock // putLock：将数据放入阻塞队列的时候需要加putLock private final ReentrantLock putLock = new ReentrantLock(); // Condition 等待条件：notFull表示 “容量未满”条件 // 当阻塞队列容量满了，放不下新的元素，此时notFull条件不满足 // 这个时候将数据放入阻塞队列的线程就会被阻塞住，等待notFull条件满足 private final Condition notFull = putLock.newCondition(); }
3.2 默认的无参构造函数
如果使用默认的无参构造函数，这里给capacity设置的容量为Integer.MAX_VALUE，这个数组很大，这也是为什么使用默认构造函数下，LinkedBlockingQueue被叫做无界阻塞队列的原因。
public LinkedBlockingQueue() { this(Integer.MAX_VALUE); }
根据上面说明的LinkedBlockingQueue的内部结构和属性，可以得出下面的这样一个图：
（1）首先有一个存储元素的链表，这个链表有头、尾两个指针
（2）然后有一个属性capacity，表示当前阻塞队列的最大容量；有一个属性count表示当前阻塞队列的大小，也就是元素的个数
（3）有一个插入锁putLock，对应的锁种类是ReentrantLock，还有对应一个插入条件是notFull (Condition)，还有一个notFull条件对应的等待队列
（4）同样的，对应拉取元素也有一个takeLock，等待条件是notEmpty，还有对应的等待队列
（5）使用ReentrantLock和Condition主要是为了实现线程间的阻塞和唤醒操作，这个之前讲解ReentrantLock的时候详细讲解过了
3.3 put方法
那我们继续，看一下LinkedBlockingQueue的put方法源码。
public void put(E e) throws InterruptedException { // 如果要插入元素是null，抛出异常 if (e == null) throw new NullPointerException(); int c = -1; // 创建一个Node节点，e是要存储的元素 Node<E> node = new Node<E>(e); // putLock插入锁 final ReentrantLock putLock = this.putLock; // 当前队列中元素的大小 final AtomicInteger count = this.count; // 调用lockInterruptibly获取插入锁 // 如果获取锁成功，则继续运行，获取失败则阻塞在这里 putLock.lockInterruptibly(); try { // 最重要的来了：这里会判断当前队列count大小是否达到了阻塞队列容量上限 while (count.get() == capacity) { // 如果达到了上限，说明队列满了，notFull条件不满足 // 此时就需要调用notFull.await方法进入等待队列沉睡 // 等待队列有空余位置的时候，别的线程将它唤醒 notFull.await(); } // 如果队列未满，此时调用enqueue将node节点插入链表中 enqueue(node); c = count.getAndIncrement(); // c+1 表示插入元素之后，当前队列的大小 if (c + 1 < capacity) // 如果c+1 < capacity表示阻塞队列还未满，还有位置可以插入 // 此时调用notFull.singal方法唤醒一个因为队列满了插入阻塞的一个线程 // 告诉它，嘿哥们，醒醒了，阻塞队列有空余的位置提供你插入元素了 notFull.signal(); } finally { // 到这里插入元素操作完成，不管成功还是失败都要释放插入锁了 putLock.unlock(); } // 如果c==0，表示插入元素前阻塞队列没有元素，此时可能有别的线程取元素的时候被阻塞了 // 而现在队列的大小为 c+1 > 0 表示队列有元素了 // 此时需要调用唤醒取元素阻塞的线程，现在我新加入了一个元素，可以醒来取元素了 if (c == 0) signalNotEmpty(); }
3.3.1 enqueue方法
private void enqueue(Node<E> node) { // 插入元素并将尾指针指向插入的元素 last = last.next = node; }
3.3.2 singalNotEmpty方法
private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; // 调用takeLock.lock方法获取takeLock takeLock.lock(); try { // 之前可能有线程拉取元素的时候，由于队列是空，被阻塞了 // 此时调用notEmpty.singal唤醒这个线程 // 告诉它此时队列不是空的，有元素可取了，notEmpty条件满足了，可以醒来取数据了 notEmpty.signal(); } finally { // 这里就是常规的释放锁操作了 takeLock.unlock(); } }
上面的put方法，我们画个图来理解一下：
总结一下put方法插入的流程：
（1）首先线程A要调用put方法往阻塞队列插入一个元素，首先就需要获取putLock互斥锁。
（2）获取锁成功之后，判断当前队列大小count 是否等于 capacity，也就是是否还有位置可以插入
（3）如果队列满了，没有空间插入了，说明notFull条件不满足，此时调用notFull.await方法阻塞等待
（4）如果队列未满，此时直接往队列尾部插入一个元素，插入完成之后释放锁
（5）然后插入完成之后判断插入前的队列大小c == 0 是否满足。如果c == 0说明插入前队列是空的，有可能别的线程获取元素的时候因为队列是空被阻塞了，需要调用notEmpty.singal方法唤醒一个沉睡的线程，告诉它醒醒了，可以取元素了。
3.4 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 { // 当队列元素个数是0 while (count.get() == 0) { // 说明此时队列没有元素可以获取，notEmpty非空条件不满足 // 此时调用notEmpty.await方法进入等待队列阻塞等待 notEmpty.await(); } // 如果队列非空，此时出队一个元素 x = dequeue(); // 元素个数减少1，c是出队前的队列大小，比此时真实的大小多1 c = count.getAndDecrement(); // 如果此时c>1，说明队列元素至少有2个 if (c > 1) // 此时队列出队一个元素之后，至少还有一个元素 // 这个时候调用notEmpty唤醒一个因为队列没有元素而阻塞沉睡的线程 // 告诉它队列还有元素，赶紧醒来获取吧 notEmpty.signal(); } finally { // 释放takeLock takeLock.unlock(); } // 如果获取前c==capacity，此时出队一个元素则 c == capacity - 1 // 则此时队列还有位置插入 if (c == capacity) // 唤醒因为队列满则被阻塞的线程，告诉它有位置空余了 // 赶紧醒来插入数据了 signalNotFull(); // 返回出队的元素 return x; }
还是一样我们接着上面的put方法的图，画一下take方法的（紫色部分）：
总结一下take方法的源码步骤：
（1）线程B调用take方法从阻塞队列获取元素，首先需要进行调用takeLock.lock进行加锁
（2）加锁成功之后，判断当前队列是否有元素可以获取，即判断 count == 0
（3）如果count == 0表示当前队列是空的，没有元素获取，notEmpty非空条件不满足，此时调用notEmpty.await方法进入等待队列，等待有元素的时候别的线程将它唤醒
（4）如果count != 0 说明队列有元素可以获取，此时出队一个元素
（5）然后判断 c （获取元素之前队列的容量），如果 c == capacity说明出队前，队列是满的，可能有部分线程因为队列满了，插入操作被阻塞住了，所以需要调用notFull.singal方法将它唤醒。告诉它此时有空余位置可以插入了，醒醒了，去插入数据吧。
3.5 其它方法
接下来我们看看LinkedBlockingQueue提供的其它方法:
3.5.1 offer(E e) 插入失败非阻塞方法
public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; // 判断容量满了，直接返回false if (count.get() == capacity) return false; int c = -1; // 创建一个新的node节点 Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; // 获取插入锁 putLock.lock(); try { // 如果容量未满 if (count.get() < capacity) { // 插入队列 enqueue(node); // c = 插入前的队列容量 // 此时队列的容量 + 1 c = count.getAndIncrement(); // c + 1 是当前队列的容量，如果小于capacity表示队列未满 // 唤醒一下因为容量满插入失败的线程，告诉它有空余位置可以插入，醒醒了 if (c + 1 < capacity) notFull.signal(); } } finally { // 释放锁 putLock.unlock(); } // c = 插入前的容量，如果等于0，可能有部分拉取数据的线程之前被阻塞 // 此时唤醒一下因为拉取数据被阻塞的线程，队列有元素了，可以获取了 if (c == 0) signalNotEmpty(); return c >= 0; }
3.5.2 offer(E e, long timeout, TimeUnit unit) 允许阻塞一段时间
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { // 如果插入元素是null，抛出异常 if (e == null) throw new NullPointerException(); // 将时间转化成纳秒 long nanos = unit.toNanos(timeout); int c = -1; final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; // 插入前进行加锁 putLock.lockInterruptibly(); try { // 注意，这里while是核心逻辑，如果count == capacity说明容量满了 while (count.get() == capacity) { // 只要容量满了，进入这里，判断nacos剩余可阻塞时间是否大于0 // 如果小于等于0，说明到时间了，返回插入失败了 if (nanos <= 0) return false; // 这里就调用notFull.awaitNacos方法最多阻塞nacos的时间 // 超过了这个时间线程会自动苏醒 nanos = notFull.awaitNanos(nanos); } // 如果走到了这里，说明未超时，并且队列容量未满 // 此时就插入一个元素 enqueue(new Node<E>(e)); // c为插入前队列的大小 // 此时队列大小count自增1 c = count.getAndIncrement(); // c+1 表示当前队列大小，如果小于capacity说明队列为满 if (c + 1 < capacity) // 此时就可以唤醒一个因为插入而被阻塞的线程，还有位置，哥们醒醒 notFull.signal(); } finally { // 常规操作，释放锁 putLock.unlock(); } if (c == 0) // 如果c == 0说明插入前队列是空的，可能有一部分取数据线程被阻塞了 // 此时插入成功，队列有元素了，该唤醒取数据的线程，让它们来取数据了 signalNotEmpty(); return true; }
3.5.3 poll() 取数，队列是空时非阻塞
public E poll() { final AtomicInteger count = this.count; // 如果当前队列是空，直接返回false if (count.get() == 0) return null; E x = null; int c = -1; final ReentrantLock takeLock = this.takeLock; // 获取数据前进行加锁 takeLock.lock(); try { // 如果count > 0 说明队列有元素 if (count.get() > 0) { // 此时出队一个元素 x = dequeue(); // c为出队前队列大小 // 此时队列大小减少1 c = count.getAndDecrement(); if (c > 1) // c>1说明c至少是2，此时出队一个元素，队列至少还有一个 // 唤醒一个因为队列是空，取数时被阻塞的线程，告诉它可以取数据了 notEmpty.signal(); } } finally { // 释放锁 takeLock.unlock(); } if (c == capacity) // 出队前队列的大小，如果等于capacity，表示取数前队列是满的 // 可能之前有一部分插入线程被阻塞了，此时唤醒一个线程，空了一个位置，可以插入了 signalNotFull(); return x; }
3.5.4 poll(long timeout, TimeUnit 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.lockInterruptibly(); try { // 如果当前队列是空格的，会一直在while循环体中 while (count.get() == 0) { // 判断当前剩余可阻塞时间nanos是否大于0 // 如果小于0，说明超时了都没取数成功，返回null if (nanos <= 0) return null; // 调用notEmpty.awaitNanos方法最多阻塞nanos时间 nanos = notEmpty.awaitNanos(nanos); } // 走到这里，说明未超时，并且队列不是空，有可取的数据 // 此时就出队一个元素 x = dequeue(); // c为出队前阻塞队列的大小 // 此时阻塞队列大小减少1 c = count.getAndDecrement(); if (c > 1) // c>1说明c至少是2，此时出队一个元素，队列至少还有一个 // 唤醒一个因为队列是空，取数时被阻塞的线程，告诉它可以取数据了 notEmpty.signal(); } finally { // 释放锁 takeLock.unlock(); } if (c == capacity) // 出队前队列的大小，如果等于capacity，表示取数前队列是满的 // 可能之前有一部分插入线程被阻塞了，此时唤醒一个线程，空了一个位置，可以插入了 signalNotFull(); return x; }
3.5.5 其它方法小结
上面的就是LinkedBlockingQueue阻塞队列里面的其它方法了：
（1）包含非阻塞的插入方法offer(E e)，非阻塞的取数方法poll()
（2）包含允许阻塞时间自定义的插入方法 offer(E e,long timeout, TimeUnit) ，允许阻塞时间自定义的取数方法poll(long timeout,TimeUnit unit)
这些方法里面的实现跟之前的put和take方法里面的步骤和机制都非常相似，这里就不再画图讲解了。
4 小结
到这里，LinkedBlockingQueue我们就看的差不多了，它内部有个链表，两个互斥重入锁以及两个条件唤醒，有理解不对的地方欢迎指正哈。