深入理解ArrayBlockingQueue

╰+哭是因爲堅強的太久メ 2022-05-30 12:24 127阅读 0赞

# 说明 #

通过分析源码，了解ArrayBlockingQueue的特性。本文基于JDK1.7源码进行分析，在此进行记录和总结，若有错误，欢迎指正

# 正文 #

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

*  一个由**数组支持的有界队列**，此队列按\*\*FIFO（先进先出）\*\*原则对元素进行排序。
 *  新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素
 *  这是一个简单的“有界缓存区”，**一旦创建，就不能在增加其容量**
 *  在向已满队列中添加元素会导致操作阻塞，从空队列中提取元素也将导致阻塞
 *  此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。默认情况下，不保证是这种排序的。然而通过**将公平性（fairness）设置为true**，而构造的**队列允许按照FIFO顺序访问线程**。**公平性通常会降低吞吐量，但也减少了可变性和避免了“不平衡性”**

## ArrayBlockingQueue ##

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {

通过源码可知，ArrayBlockingQueue继承自AbstractQueue,实现了BlockingQueue,Serializable接口

### BlockingQueue接口 ###

BlockingQueue接口继承了超级接口Queue,支持两个附加操作的Queue:获取元素时等待队列变为非空，以及存储元素时等待空间变得可用  
它的方法以四种形式出现，对于不能立即满足但可能将来满足的操作，有四种处理方式：

1.  抛出异常：add(e) remove() element()
2.  返回一个特殊值（null或false,具体取决于操作）： offer(e) poll() peek()
3.  操作成功前，无限期地阻塞：put(e) take()
4.  阻塞给定的时间：offer(e,time,unit) poll(time,unit)

它有如下特点：

*  **不接受null元素**，当添加一个null元素时，会抛出NullPointerException异常
 *  可以限定容量
 *  实现主要用于生产者-消费者队列
 *  **线程安全的**，所有排队方法都可以使用内部锁或其他形式的并发控制来自动达到它们的目的

## ArrayBlockingQueue的属性 ##

final Object[] items;  //用于存储元素的数组
    
        /** items index for next take, poll, peek or remove */
        int takeIndex;  // 下一个取出元素的坐标
    
        /** items index for next put, offer, or add */
        int putIndex; // 可以添加元素的坐标
    
        int count;  // 队列中元素的数量
    
        /*
         * Concurrency control uses the classic two-condition algorithm
         * found in any textbook.
         */
    
        /** Main lock guarding all access */
        final ReentrantLock lock;           // 所有的入队出队都是用同一锁
        /** Condition for waiting takes */
        private final Condition notEmpty;  // 由lock创键的Condition notEmpty 非空即当取出元素时使用
        /** Condition for waiting puts */
        private final Condition notFull;  // 同样由lock创建，添加时使用

## 构造函数 ##

public ArrayBlockingQueue(int capacity) {
            this(capacity, false);
        }

public ArrayBlockingQueue(int capacity, boolean fair) {
            if (capacity <= 0)
                throw new IllegalArgumentException();
            this.items = new Object[capacity];
            lock = new ReentrantLock(fair);
            notEmpty = lock.newCondition();
            notFull =  lock.newCondition();
        }

通过ArrayBlockingQueue的这两个构造函数可以看出：**在创建时，必须指定容量capacity,在默认时，采用非公平的策略**

## takeIndex和putIndex变化函数 ##

final int inc(int i) {
            return (++i == items.length) ? 0 : i;
        }

final int dec(int i) {
            return ((i == 0) ? items.length : i) - 1;
        }

通过这两个函数可以看出，数组坐标是循环的

## ArrayBlockingQueue添加方法 ##

### offer(e) ###

public boolean offer(E e) {
            checkNotNull(e);  // 先判断添加元素是否为null
            final ReentrantLock lock = this.lock;
            lock.lock();  // 加锁
            try {
                if (count == items.length)
                    return false;  // 如果队列已满，直接返回false
                else {
                    insert(e);   //否则调用insert方法，返回true
                    return true;
                }
            } finally {
                lock.unlock();
            }
        }

### put(e) ###

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);  // 否则调用insert方法
            } finally {
                lock.unlock();
            }
        }

### offer(e,timeout,unit) ###

public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {
    
            checkNotNull(e);
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == items.length) {
                    if (nanos <= 0)    //如果指定时间内操作未成功，则返回false
                        return false;
                    nanos = notFull.awaitNanos(nanos); // 如果队列已满，则阻塞给定时间
                }
                insert(e); // 同样调用insert方法
                return true;
            } finally {
                lock.unlock();
            }
        }

通过以上三种添加方法，发现在方法的内部都调用了insert方法

private void insert(E x) {
            items[putIndex] = x; 
            putIndex = inc(putIndex); // 添加后 坐标加1
            ++count;
            notEmpty.signal(); // 取出线程被唤醒
        }

## ArrayBlockingQueue取出方法 ##

### poll() ###

public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : extract(); // 如果队列为空返回null，否则调用extract()
            } finally {
                lock.unlock();
            }
        }

### take() ###

public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0)
                    notEmpty.await(); // 当队列为空时，阻塞等待
                return extract();
            } finally {
                lock.unlock();
            }
        }

### poll(timeout,unit) ###

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0) {
                    if (nanos <= 0)
                        return null; // 超过时间 操作未成功 返回null
                    nanos = notEmpty.awaitNanos(nanos);//当队列为空时，阻塞给定时间
                }
                return extract();
            } finally {
                lock.unlock();
            }
        }

通过以上三种从队头取出元素方法中可以看出，都调用了extract()方法

private E extract() {
            final Object[] items = this.items;
            E x = this.<E>cast(items[takeIndex]);
            items[takeIndex] = null; //取出时，将该位置置为null
            takeIndex = inc(takeIndex); //取出坐标加一
            --count;
            notFull.signal(); //唤醒添加线程
            return x;
        }

### peek() ###

该方法只返回队头元素，不从队列中删除

public E peek() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : itemAt(takeIndex); //当队列为空返回null
            } finally {
                lock.unlock();
            }
        }

### remove(Object o) ###

删除某个指定的元素，先找到该元素在数组中的坐标，然后使用removeAt(i)方法删除

public boolean remove(Object o) {
            if (o == null) return false;
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {//通过循环遍历，找出指定元素所在位置
                for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) {
                    if (o.equals(items[i])) {
                        removeAt(i); // 找到坐标后，调用removeAt(i)方法
                        return true; //成功删除 返回true
                    }
                }
                return false; // 总则返回false
            } finally {
                lock.unlock();
            }
        }

void removeAt(int i) {
            final Object[] items = this.items;
            // if removing front item, just advance
            if (i == takeIndex) { //如果i刚好为取出坐标，直接置为null
                items[takeIndex] = null;
                takeIndex = inc(takeIndex);
            } else {
                // slide over all others up through putIndex.
                for (;;) { // 否则将删除位置之后的元素都向前移一位
                    int nexti = inc(i);
                    if (nexti != putIndex) {
                        items[i] = items[nexti];
                        i = nexti;
                    } else {
                        items[i] = null;
                        putIndex = i;
                        break;
                    }
                }
            }
            --count;
            notFull.signal(); //唤醒添加线程
        }

## 返回队列大小及剩余容量 ##

public int size() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }

public int remainingCapacity() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return items.length - count; // 剩余容量为数组长度减去当前元素数量
            } finally {
                lock.unlock();
            }
        }

通过介绍以上方法，我们发现在ArrayBlockingQueue中所有方法都是使用同一个锁this.lock来保障线程安全