Java并发容器之有界阻塞队列ArrayBlockingQueue

喜欢ヅ旅行 2022-05-25 03:46 224阅读 0赞

在这篇文章中，我们来看下另一个线程安全的队列容器，那就是ArrayBlockingQueue。

它是基于数组实现的有界阻塞队列。以JDK1.7为参考，ArrayBlockingQueue的类图结构如下所示：

![70][]

在ArrayBlockingQueue内部采用数组来保存队列元素，同时维护两个变量putIndex和takeIndex来标识插入元素和获取元素的下标位置，同时定义了count变量用来统计队列元素个数。如下所示：

/** The queued items */
        final Object[] items;
    
        /** items index for next take, poll, peek or remove */
        int takeIndex;
    
        /** items index for next put, offer, or add */
        int putIndex;
    
        /** Number of elements in the queue */
        int count;

之所以这里定义的变量没有使用volatile关键字修饰，是因为这些变量的访问都是在锁的保护下进行的，不存在可见性问题。

为了保证线程安全，在队列的出队和入队过程中都是在锁的保护下完成的，在创建这个阻塞队列时会创建一个ReentrantLock锁，并且基于该锁创建两个Condition条件变量对象用于出入队列的同步。

/** Main lock guarding all access */
        final ReentrantLock lock;
        /** Condition for waiting takes */
        private final Condition notEmpty;
        /** Condition for waiting puts */
        private final Condition notFull;

**Put操作**

源码如下所示：

/**
         * Inserts the specified element at the tail of this queue, waiting
         * for space to become available if the queue is full.
         *
         * @throws InterruptedException {@inheritDoc}
         * @throws NullPointerException {@inheritDoc}
         */
        public void put(E e) throws InterruptedException {
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == items.length)
                    notFull.await();
                insert(e);
            } finally {
                lock.unlock();
            }
        }

/**
         * Inserts element at current put position, advances, and signals.
         * Call only when holding lock.
         */
        private void insert(E x) {
            items[putIndex] = x;
            putIndex = inc(putIndex);
            ++count;
            notEmpty.signal();
        }

在入队前先获取锁，之后判断队列是否已满，如果是，则等待notFull条件的signal()信号，之后进行插入操作；如果队列不满，则直接插入对象到列队中。至于这里为什么要使用lock.lockInterruptibly而不是lock.lock有明白的可以在评论区补充说明，谢谢。

**Offer操作**

该操作用于在队列尾部插入元素，如果队列满了，则直接返回false，否则将对象入队，并返回true。

/**
         * Inserts the specified element at the tail of this queue if it is
         * possible to do so immediately without exceeding the queue's capacity,
         * returning {@code true} upon success and {@code false} if this queue
         * is full.  This method is generally preferable to method {@link #add},
         * which can fail to insert an element only by throwing an exception.
         *
         * @throws NullPointerException if the specified element is null
         */
        public boolean offer(E e) {
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                if (count == items.length)
                    return false;
                else {
                    insert(e);
                    return true;
                }
            } finally {
                lock.unlock();
            }
        }

这个方法相比于Put操作都是在获取锁之后进行的，但是在获取锁之后，如果队列满了就直接返回失败，而如果队列不满则直接入队并返回成功，没有阻塞。与offer方法相同作用的还有add方法，但是add方法在入队失败后不是返回false，而是抛出异常。

**take操作**

take操作用于从队列头获取并返回元素，如果队列为空，则阻塞直到队列中有元素。

public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0)
                    notEmpty.await();
                return extract();
            } finally {
                lock.unlock();
            }
        }

private E extract() {
            final Object[] items = this.items;
            E x = this.<E>cast(items[takeIndex]);
            items[takeIndex] = null;
            takeIndex = inc(takeIndex);
            --count;
            notFull.signal();
            return x;
        }

在上面take方法中，当队列为空时，线程阻塞在notEmpty.await()处，直到接受到队列不为空的信号notEmpty.signal()。在从队列头取出元素之后，队列明显不满，这时候调用notFull.signal()通知那些等待向队列中插入元素的线程。

**Poll操作**

poll操作也是用于从队列头获取并移除元素，如果队列为空，则直接返回null，而不是进行阻塞等待。

public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : extract();
            } finally {
                lock.unlock();
            }
        }

**Peek操作**

peek操作用于返回队列头元素，但是并不移除该元素，如果队列中没有元素，则返回null。

public E peek() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : itemAt(takeIndex);
            } finally {
                lock.unlock();
            }
        }

/**
         * Returns item at index i.
         */
        final E itemAt(int i) {
            return this.<E>cast(items[i]);
        }

**总结**

ArrayBlockingQueue通过使用全局独占锁实现同时只能有一个线程进行入队或者出队操作，这个锁的粒度比较大，有点类似在方法上添加synchronized的意味。

考虑到队列的操作（入队、出队、获取元素）涉及到多种实现，有返回特殊值的，有抛出异常的，还有阻塞的，下面对这些方法做个总结，方便大家记忆。

<table>  队列操作  
 <tbody> 
  <tr> 
   <th>&nbsp;</th> 
   <th>特殊值</th> 
   <th>抛出异常</th> 
   <th>阻塞</th> 
  </tr> 
 </tbody> 
 <tbody> 
  <tr> 
   <th>添加元素</th> 
   <td>offer</td> 
   <td>add</td> 
   <td>put</td> 
  </tr> 
  <tr> 
   <th>移除元素</th> 
   <td>poll</td> 
   <td>remove</td> 
   <td>take</td> 
  </tr> 
  <tr> 
   <th>获取元素</th> 
   <td>peek</td> 
   <td>element</td> 
   <td>不支持</td> 
  </tr> 
 </tbody> 
</table>

感谢大家的阅读，如果有对Java编程、中间件、数据库、及各种开源框架感兴趣，欢迎关注我的博客和头条号（源码帝国），博客和头条号后期将定期提供一些相关技术文章供大家一起讨论学习，谢谢。

如果觉得文章对您有帮助，欢迎给我打赏，一毛不嫌少，一百不嫌多，^\_^谢谢。

![20180312232547620][]

[70]: /images/20220525/4595953fee1e4ea2a43b87e7e21697bc.png
[20180312232547620]: /images/20220525/358b7119b7ce40839e23847f3d6c61dc.png