jdk源码解析三之ArrayBlockingQueue

我会带着你远行 2023-02-20 12:08 19阅读 0赞

### 文章目录 ###

*   *  ArrayBlockingQueue
     *   *  put
         *  offer
         *  take
         *  remove
         *  总结

## ArrayBlockingQueue ##

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

### put ###

public void put(E e) throws InterruptedException {
        
            //value不允许为null
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            //中断锁
            lock.lockInterruptibly();
            try {
        
                //当前count超过生产的容量上限,则等待
                while (count == items.length)
                    notFull.await();
                enqueue(e);
            } finally {
        
                //释放锁
                lock.unlock();
            }
        }
    
    private void enqueue(E x) {
        
            // assert lock.getHoldCount() == 1;
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            items[putIndex] = x;
            //当前个数超过下标,则回滚为0下标
            if (++putIndex == items.length)
                putIndex = 0;
            //当前可进行消息个数+1
            count++;
            //唤醒消费线程进行消费
            notEmpty.signal();
        }

### offer ###

public boolean offer(E e) {
        
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
        
                //当前消息个数超过上限,则直接返回
                if (count == items.length)
                    return false;
                else {
        
                    enqueue(e);
                    return true;
                }
            } finally {
        
                lock.unlock();
            }
        }

### take ###

public E take() throws InterruptedException {
        
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
        
                //当前没有消息,则等待
                while (count == 0)
                    notEmpty.await();
                //消费消息
                return dequeue();
            } finally {
        
                lock.unlock();
            }
        }
    
        private E dequeue() {
        
            // assert lock.getHoldCount() == 1;
            // assert items[takeIndex] != null;
            final Object[] items = this.items;
            @SuppressWarnings("unchecked")
                    //获取消息
            E x = (E) items[takeIndex];
            //清空消息
            items[takeIndex] = null;
            //当前消费的索引为数组长度,则从0开始消费消息
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;
            //迭代器冲突处理
            if (itrs != null)
                itrs.elementDequeued();
            //唤醒生产线程,生产消息
            notFull.signal();
            return x;
        }
    
            void elementDequeued() {
        
                // assert lock.getHoldCount() == 1;
                //消费个数为0，则清空迭代器值
                if (count == 0)
                    queueIsEmpty();
                else if (takeIndex == 0)
                    //消费的索引回滚为0处理
                    takeIndexWrapped();
            }

### remove ###

public boolean remove(Object o) {
        
            if (o == null) return false;
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
        
                if (count > 0) {
        
                    final int putIndex = this.putIndex;
                    int i = takeIndex;
                    //遍历takeIndex ~ putIndex之间的数据,如果涉及到边界问题,则从0开始查找
                    do {
        
                        //匹配则删除
                        if (o.equals(items[i])) {
        
                            removeAt(i);
                            return true;
                        }
                        //边界检测.
                        if (++i == items.length)
                            i = 0;
                    } while (i != putIndex);
                }
                return false;
            } finally {
        
                lock.unlock();
            }
        }
    
     void removeAt(final int removeIndex) {
        
            // assert lock.getHoldCount() == 1;
            // assert items[removeIndex] != null;
            // assert removeIndex >= 0 && removeIndex < items.length;
            final Object[] items = this.items;
            //如果删除的消息和目前正take索引相同
            if (removeIndex == takeIndex) {
        
                // removing front item; just advance
                //清空当前take消息
                items[takeIndex] = null;
                //takeIndex偏移到下一个
                if (++takeIndex == items.length)
                    takeIndex = 0;
                count--;
                //迭代器冲突处理
                if (itrs != null)
                    itrs.elementDequeued();
            } else {
        
                // an "interior" remove
    
                // slide over all others up through putIndex.
                final int putIndex = this.putIndex;
                for (int i = removeIndex;;) {
        
                    int next = i + 1;
                    //达到数组上线,索引设为0,开始遍历
                    if (next == items.length)
                        next = 0;
                    //未到边界,则偏移到下一个数组值
                    if (next != putIndex) {
        
                        items[i] = items[next];
                        i = next;
                    } else {
        
                        //到了边界,设置当前items[putIndex]值为null,且更新putIndex
                        items[i] = null;
                        this.putIndex = i;
                        break;
                    }
                }
                //当前消息个数-1
                count--;
                //迭代器冲突处理
                if (itrs != null)
                    itrs.removedAt(removeIndex);
            }
            notFull.signal();
        }

### 总结 ###

有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。  
初始化必须设置容量,value不允许为空  
全局一个锁处理,相对并发比较低.  
因为底层数组,所以修改查询快.