并发编程（九）—— Java 并发队列 BlockingQueue 实现之 LinkedBlockingQueue 源码分析

linkedblockingqueue

在看源码之前，通过查询api发现对linkedblockingqueue特点的简单介绍：

1、linkedblockingqueue是一个由链表实现的有界队列阻塞队列。
2、新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素
3、大小默认值为integer.max_value，所以我们在使用linkedblockingqueue时建议手动传值，为其提供我们所需的大小，避免队列过大造成机器负载或者内存爆满等情况。

4、链接队列的吞吐量通常要高于基于数组的对列（arrayblockingqueue）,但是在大多数并发应用程序中，其可预知的性能要低

源码剖析

通常情况下，我们建议创建指定大小的linkedblockingqueue阻塞队列。在正常情况下，链接队列的吞吐量要高于基于数组的队列（arrayblockingqueue），因为其内部实现添加和删除操作使用的两个reenterlock来控制并发执行，而arrayblockingqueue内部只是使用一个reenterlock控制并发，因此linkedblockingqueue的吞吐量要高于arrayblockingqueue。

其构造函数如下：

 1 //默认大小为integer.max_value
 2 public linkedblockingqueue() {
 3        this(integer.max_value);
 4 }
 5 
 6 //创建指定大小为capacity的阻塞队列
 7 public linkedblockingqueue(int capacity) {
 8      if (capacity <= 0) throw new illegalargumentexception();
 9      this.capacity = capacity;
10      last = head = new node<e>(null);
11 }

我们先看看linkedblockingqueue的内部成员变量：

 1 public class linkedblockingqueue<e> extends abstractqueue<e>
 2         implements blockingqueue<e>, java.io.serializable {
 3 
 4     /**
 5      * 节点类，用于存储数据
 6      */
 7     static class node<e> {
 8         e item;
 9 
10         /**
11          * one of:
12          * - the real successor node
13          * - this node, meaning the successor is head.next
14          * - null, meaning there is no successor (this is the last node)
15          */
16         node<e> next;
17 
18         node(e x) { item = x; }
19     }
20 
21     /** 阻塞队列的大小，默认为integer.max_value */
22     private final int capacity;
23 
24     /** 当前阻塞队列中的元素个数 */
25     private final atomicinteger count = new atomicinteger();
26 
27     /**
28      * 阻塞队列的头结点
29      */
30     transient node<e> head;
31 
32     /**
33      * 阻塞队列的尾节点
34      */
35     private transient node<e> last;
36 
37     /** 获取并移除元素时使用的锁，如take, poll, etc */
38     private final reentrantlock takelock = new reentrantlock();
39 
40     /** notempty条件对象，当队列没有数据时用于挂起执行删除的线程 */
41     private final condition notempty = takelock.newcondition();
42 
43     /** 添加元素时使用的锁如 put, offer, etc */
44     private final reentrantlock putlock = new reentrantlock();
45 
46     /** notfull条件对象，当队列数据已满时用于挂起执行添加的线程 */
47     private final condition notfull = putlock.newcondition();
48 
49 }

与arrayblockingqueue不同的是，linkedblockingqueue内部分别使用了takelock 和 putlock 对并发进行控制，也就是说，添加和删除操作并不是互斥操作，可以同时进行，这样也就可以大大提高吞吐量。这里再次强调如果没有给linkedblockingqueue指定容量大小，其默认值将是integer.max_value，如果存在添加速度大于删除速度时候，有可能会内存溢出，这点在使用前希望慎重考虑。

这里用了两个锁，两个 condition，简单介绍如下：

takelock 和 notempty 怎么搭配：如果要获取（take）一个元素，需要获取 takelock 锁，但是获取了锁还不够，如果队列此时为空，还需要队列不为空（notempty）这个条件（condition）。

putlock 需要和 notfull 搭配：如果要插入（put）一个元素，需要获取 putlock 锁，但是获取了锁还不够，如果队列此时已满，还需要队列不是满的（notfull）这个条件（condition）。

 

下面我们看看其其内部添加过程和删除过程是如何实现的。

添加方法的实现原理

对于添加方法，主要指的是add，offer以及put，这里先看看add方法和offer方法的实现

1 public boolean add(e e) {
2      if (offer(e))
3          return true;
4      else
5          throw new illegalstateexception("queue full");
6 }

从源码可以看出，add方法间接调用的是offer方法，如果add方法添加失败将抛出illegalstateexception异常，添加成功则返回true，那么下面我们直接看看offer的相关方法实现

 1 public boolean offer(e e) {
 2      //添加元素为null直接抛出异常
 3      if (e == null) throw new nullpointerexception();
 4       //获取队列的个数
 5       final atomicinteger count = this.count;
 6       //判断队列是否已满
 7       if (count.get() == capacity)
 8           return false;
 9       int c = -1;
10       //构建节点
11       node<e> node = new node<e>(e);
12       final reentrantlock putlock = this.putlock;
13       //此时可能有多个线程都在执行添加操作，抢占锁
14       //没有抢占到锁的线程会加入到putlock的阻塞队列，并且挂起
15       putlock.lock();
16       try {
17           //再次判断队列是否已满，考虑并发情况
18           if (count.get() < capacity) {
19               enqueue(node);//添加元素
20               c = count.getandincrement();//拿到当前未添加新元素时的队列长度
21               //如果容量还没满
22               if (c + 1 < capacity)
23                   //此时会唤醒上面第15行处添加到putlock的阻塞队列的线程，被唤醒的线程满足条件后接着唤醒下一个，直到队列被添加满
24                   notfull.signal();//唤醒下一个添加线程，执行添加操作
25           }
26       } finally {
27           putlock.unlock();
28       }
29       // 由于存在添加锁和消费锁，而消费锁和添加锁都会持续唤醒等到线程，因此count肯定会变化。
30       //这里的if条件表示如果队列中还有1条数据
31       if (c == 0) 
32         signalnotempty();//如果还存在数据那么就唤醒消费锁
33     return c >= 0; // 添加成功返回true，否则返回false
34 }
35 
36 //入队操作
37 private void enqueue(node<e> node) {
38      //队列尾节点指向新的node节点
39      last = last.next = node;
40 }
41 
42 //signalnotempty方法
43 private void signalnotempty() {
44       final reentrantlock takelock = this.takelock;
45       takelock.lock();
46           //唤醒获取并删除元素的线程
47           notempty.signal();
48       } finally {
49           takelock.unlock();
50       }
51 }

 

这里的offer()方法做了两件事，第一件事是判断队列是否满，满了就直接释放锁，没满就将节点封装成node入队，然后再次判断队列添加完成后是否已满，不满就继续唤醒等到在条件对象notfull上的添加线程。第二件事是，判断是否需要唤醒等到在notempty条件对象上的消费线程。这里我们可能会有点疑惑，为什么添加完成后是继续唤醒在条件对象notfull上的添加线程而不是像arrayblockingqueue那样直接唤醒notempty条件对象上的消费线程？而又为什么要当if (c == 0)时才去唤醒消费线程呢？

　　唤醒添加线程的原因，在添加新元素完成后，会判断队列是否已满，不满就继续唤醒在条件对象notfull上的添加线程，这点与前面分析的arrayblockingqueue很不相同，在arrayblockingqueue内部完成添加操作后，会直接唤醒消费线程对元素进行获取，这是因为arrayblockingqueue只用了一个reenterlock同时对添加线程和消费线程进行控制，这样如果在添加完成后再次唤醒添加线程的话，消费线程可能永远无法执行，而对于linkedblockingqueue来说就不一样了，其内部对添加线程和消费线程分别使用了各自的reenterlock锁对并发进行控制，也就是说添加线程和消费线程是不会互斥的，所以添加锁只要管好自己的添加线程即可，添加线程自己直接唤醒自己的其他添加线程，如果没有等待的添加线程，直接结束了。如果有就直到队列元素已满才结束挂起，当然offer方法并不会挂起，而是直接结束，只有put方法才会当队列满时才执行挂起操作。注意消费线程的执行过程也是如此。这也是为什么linkedblockingqueue的吞吐量要相对大些的原因。

　　为什么要判断if (c == 0)时才去唤醒消费线程呢，这是因为消费线程一旦被唤醒是一直在消费的（前提是有数据），所以c值是一直在变化的，c值是添加完元素前队列的大小，此时c只可能是0或c>0，如果是c=0，那么说明之前消费线程已停止，条件对象上可能存在等待的消费线程，添加完数据后应该是c+1，那么有数据就直接唤醒等待消费线程，如果没有就结束啦，等待下一次的消费操作。如果c>0那么消费线程就不会被唤醒，只能等待下一个消费操作（poll、take、remove）的调用，那为什么不是条件c>0才去唤醒呢？我们要明白的是消费线程一旦被唤醒会和添加线程一样，一直不断唤醒其他消费线程，如果添加前c>0，那么很可能上一次调用的消费线程后，数据并没有被消费完，条件队列上也就不存在等待的消费线程了，还有可能，消费线程正在消费，消费队列中也有等待线程，但是消费线程消费完会自动唤醒下一个等待的消费线程，所以c>0唤醒消费线程得意义不是很大，当然如果添加线程一直添加元素，那么一直c>0，消费线程执行的换就要等待下一次调用消费操作了（poll、take、remove）。

 

接下来我们看看put方法，它是一个阻塞添加的方法：

 1 public void put(e e) throws interruptedexception {
 2     if (e == null) throw new nullpointerexception();
 3     int c = -1;
 4     node<e> node = new node(e);
 5     final reentrantlock putlock = this.putlock;
 6     final atomicinteger count = this.count;
 7     putlock.lockinterruptibly();
 8     try {
 9         // 如果队列满，等待 notfull 的条件满足。
10         while (count.get() == capacity) {
11             notfull.await();
12         }
13         // 入队
14         enqueue(node);
15         // count 原子加 1，c 还是加 1 前的值
16         c = count.getandincrement();
17         if (c + 1 < capacity)
18             notfull.signal();
19     } finally {
20         // 入队后，释放掉 putlock
21         putlock.unlock();
22     }
23     if (c == 0)
24         signalnotempty();
25 }

put 和 offer最大的区别是put方法是阻塞的，看offer 方法中的第6行，如果队列满了，则直接返回false；put方法的第10行，如果队列满了，则添加到 notfull 的等待队列中并挂起，其他的基本都一样。

 

移除方法的实现原理

我们先来看看poll()方法

 1 public e poll() {
 2          //获取当前队列的大小
 3         final atomicinteger count = this.count;
 4         if (count.get() == 0)//如果没有元素直接返回null
 5             return null;
 6         e x = null;
 7         int c = -1;
 8         final reentrantlock takelock = this.takelock;
 9         takelock.lock();
10         try {
11             //判断队列是否有数据
12             if (count.get() > 0) {
13                 //如果有，直接删除并获取该元素值
14                 x = dequeue();
15                 //当前队列大小减一
16                 c = count.getanddecrement();
17                 //如果队列未空，继续唤醒等待在条件对象notempty上的消费线程
18                 if (c > 1)
19                     notempty.signal();
20             }
21         } finally {
22             takelock.unlock();
23         }
24         //判断c是否等于capacity，这是因为如果满说明notfull条件对象上
25         //可能存在等待的添加线程
26         if (c == capacity)
27             signalnotfull();
28         return x;
29 }
30 
31 private e dequeue() {
32         node<e> h = head;//获取头结点
33         node<e> first = h.next; 获取头结的下一个节点（要删除的节点）
34         h.next = h; // help gc//自己next指向自己，即被删除
35         head = first;//更新头结点
36         e x = first.item;//获取删除节点的值
37         first.item = null;//清空数据，因为first变成头结点是不能带数据的，这样也就删除队列的带数据的第一个节点
38         return x;
39 }

poll方法也比较简单，如果队列没有数据就返回null，如果队列有数据，那么就取出来，如果队列还有数据那么唤醒等待在条件对象notempty上的消费线程。然后判断if (c == capacity)为true就唤醒添加线程，这点与前面分析if(c==0)是一样的道理。因为只有可能队列满了，notfull条件对象上才可能存在等待的添加线程。

我们再看看take()方法

 1 public e take() throws interruptedexception {
 2         e x;
 3         int c = -1;
 4         //获取当前队列大小
 5         final atomicinteger count = this.count;
 6         final reentrantlock takelock = this.takelock;
 7         takelock.lockinterruptibly();//可中断
 8         try {
 9             //如果队列没有数据，挂机当前线程到条件对象的等待队列中
10             while (count.get() == 0) {
11                 notempty.await();
12             }
13             //如果存在数据直接删除并返回该数据
14             x = dequeue();
15             c = count.getanddecrement();//队列大小减1
16             if (c > 1)
17                 notempty.signal();//还有数据就唤醒后续的消费线程
18         } finally {
19             takelock.unlock();
20         }
21         //满足条件，唤醒条件对象上等待队列中的添加线程
22         if (c == capacity)
23             signalnotfull();
24         return x;
25 }

take方法是一个可阻塞可中断的移除方法，主要做了两件事，一是，如果队列没有数据就挂起当前线程到 notempty条件对象的等待队列中一直等待，如果有数据就删除节点并返回数据项，同时唤醒后续消费线程，二是尝试唤醒条件对象notfull上等待队列中的添加线程。 到此关于remove、poll、take的实现也分析完了，其中只有take方法具备阻塞功能。

linkedblockingqueue和arrayblockingqueue迥异

对于linkedblockingqueue和arrayblockingqueue的基本使用以及内部实现原理我们已较为熟悉了，这里我们就对它们两间的区别来个小结

1.队列大小有所不同，arrayblockingqueue是有界的初始化必须指定大小，而linkedblockingqueue可以是有界的也可以是*的(integer.max_value)，对于后者而言，当添加速度大于移除速度时，在*的情况下，可能会造成内存溢出等问题。

2.数据存储容器不同，arrayblockingqueue采用的是数组作为数据存储容器，而linkedblockingqueue采用的则是以node节点作为连接对象的链表。

3.由于arrayblockingqueue采用的是数组的存储容器，因此在插入或删除元素时不会产生或销毁任何额外的对象实例，而linkedblockingqueue则会生成一个额外的node对象。这可能在长时间内需要高效并发地处理大批量数据的时，对于gc可能存在较大影响。

4.两者的实现队列添加或移除的锁不一样，arrayblockingqueue实现的队列中的锁是没有分离的，即添加操作和移除操作采用的同一个reenterlock锁，而linkedblockingqueue实现的队列中的锁是分离的，其添加采用的是putlock，移除采用的则是takelock，这样能大大提高队列的吞吐量，也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。