Java并发容器之阻塞队列LinkedBlockingQueue

￡神魔★判官ぃ 2022-05-25 03:50 275阅读 0赞

在前一篇文章《[Java并发容器之有界阻塞队列ArrayBlockingQueue][Java_ArrayBlockingQueue]》中讲到了阻塞队列，提到了ArrayBlockingQueue，在这篇文章中我们来看另一种线程安全的阻塞队列实现-LinkedBlockingQueue。

ArrayBlockingQueue底层使用数组来保存队列元素，但是LinkedBlockingQueue底层使用列表来保存队列元素。

下面我们先对LinkedBlockingQueue实现进行一个简单的说明，之后结合前一篇文章对ArrayBlockingQueue的描述，总结一下两者的区别。

**LinkedBlockingQueue的实现原理**

LinkedBlockingQueue类提供了三种构造方法用于实例化对象，分别如下所示：

/**
         * Creates a {@code LinkedBlockingQueue} with a capacity of
         * {@link Integer#MAX_VALUE}.
         */
        public LinkedBlockingQueue() {
            this(Integer.MAX_VALUE);
        }
    
        /**
         * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity.
         *
         * @param capacity the capacity of this queue
         * @throws IllegalArgumentException if {@code capacity} is not greater
         *         than zero
         */
        public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);
        }
    
        /**
         * Creates a {@code LinkedBlockingQueue} with a capacity of
         * {@link Integer#MAX_VALUE}, initially containing the elements of the
         * given collection,
         * added in traversal order of the collection's iterator.
         *
         * @param c the collection of elements to initially contain
         * @throws NullPointerException if the specified collection or any
         *         of its elements are null
         */
        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));
                    ++n;
                }
                count.set(n);
            } finally {
                putLock.unlock();
            }
        }

从上面可以看出，在默认情况下，使用Integer.MAX\_VALUE来初始化队列的容量。一般情况下我们最好设置队列的容量，否则很容易造成内存溢出问题。在第三个构造函数中，使用已有的集合对象初始化队列，内部通过先构建一个空队列然后遍历集合中的每个元素并将每个元素添加到队列中。

队列中的每个元素节点都是一个Node，定义如下，包含元素对象e以及下一个节点Node的引用，

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; }
        }

在LinkedBlockingQueue内部维护了两个Node引用，分别指向这个链表中的队列头元素和队列尾元素。

/**
         * Head of linked list.
         * Invariant: head.item == null
         */
        private transient Node<E> head;
    
        /**
         * Tail of linked list.
         * Invariant: last.next == null
         */
        private transient Node<E> last;

下面我们重点来看下是怎么添加和删除元素？

**添加元素**

在前一篇文章我们知道，对于阻塞队列来说，添加元素的方式其实有三种，分别是add、offer和put，下面分别来看一下。

（1）add

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

add底层复用的是offer方法，在offer方法成功后返回成功，如果offer方法调用失败，则抛出异常。

（2）offer

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);
            final ReentrantLock putLock = this.putLock;
            putLock.lock();
            try {
                if (count.get() < capacity) {
                    enqueue(node);
                    c = count.getAndIncrement();
                    if (c + 1 < capacity)
                        notFull.signal();
                }
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return c >= 0;
        }

这里的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）。

**移除元素**

移除元素的方法有remove、poll、take。下面来一一分析。

（1）remove方法，源码如下所示

![70][]

![70 1][]

这里的操作是在加锁的控制下进行的，并且同时加了putLock和takeLock，主要还是基于线程安全性考虑。在内部逻辑上，通过for循环遍历整个链表，找到对应的节点后将相应节点从链表中移除并返回成功。

（2）poll方法，源码如下：

![70 2][]

![70 3][]

在poll方法中首先判断链表是否为空，如果为空，直接返回null，否则进行加锁控制，并在锁的保护下从队列的头部取出第一个元素返回。如果在取出一个元素之后发现队列中还有元素，通过notEmpty.signal（）来通知其他等待在这个队列上的消费线程。

（3）take方法

源码如下：

![70 4][]

take方法是一个可阻塞可中断的移除方法，主要做了两件事，一是，如果队列没有数据就挂起当前线程到 notEmpty条件对象的等待队列中一直等待，如果有数据就删除节点并返回数据项，同时唤醒后续消费线程，二是尝试唤醒条件对象notFull上等待队列中的添加线程。

**LinkedBlockingQueue和ArrayBlockingQueue的对比**

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

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

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

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

[Java_ArrayBlockingQueue]: https://blog.csdn.net/majinggogogo/article/details/80229367
[70]: /images/20220525/7aa9d5b54c4c472d8f889500f74a6eaf.png
[70 1]: /images/20220525/0d18d81be0d2416cb3131ac9892d7ceb.png
[70 2]: /images/20220525/41568c6af0aa4c8a87b6d770ac15fff1.png
[70 3]: /images/20220525/b122fbe640bf4534963f8a243cbd4de7.png
[70 4]: /images/20220525/161daaff41ad4b45b69398009decea80.png