欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

基于数组的有界阻塞队列ArrayBlockingQueue源码分析

程序员文章站 2022-07-02 09:59:07
...

一:功能介绍
         基于数组的有界阻塞队列,基于FIFO的存储模式,支持公平非公平锁。
二:源码分析

  //数组
    final Object[] items;
    //出队索引
    int takeIndex;
    //入队索引
    int putIndex;
    //队列大小
    int count;
    //可重入锁
    final ReentrantLock lock;
    //等待通知条件
    private final Condition notEmpty;
    //等待通知条件
    private final Condition notFull;

  构造函数

public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        //初始化数组容量
        this.items = new Object[capacity];
        //内容采用可重入锁ReentrantLock实现,支持公平非公平选择
        lock = new ReentrantLock(fair);
        //阻塞队列,等待条件
        notEmpty = lock.newCondition();
        //阻塞队列,等待条件
        notFull =  lock.newCondition();
   }

   入队操作

 public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        //可中断获取锁,如果出现了interrupted,不用一直阻塞
        lock.lockInterruptibly();
        try {
            //如果队列已满
            while (count == items.length)
                //入队线程阻塞
                notFull.await();
            //插入数据
            insert(e);
        } finally {
            lock.unlock();
        }
    }
    private void insert(E x) {
        //将新的数据赋值在数组的某一个索引处
        items[putIndex] = x;
        //重新赋值putIndex,设置下一个被取出元素的索引
        putIndex = inc(putIndex);
        //队列大小+1
        ++count;
        //唤醒take线程
        notEmpty.signal();
    }
     final int inc(int i) {
        //如果队列满了,重新初始化为0
        return (++i == items.length) ? 0 : i;
    }

    出队操作

 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;
        //获取takeIndex处的元素
        E x = this.<E>cast(items[takeIndex]);
        //置空takeIndex处的元素,引用不存在,便于GC,释放内存
        items[takeIndex] = null;
        //重新赋值takeIndex,设置下一个被取出的元素
        takeIndex = inc(takeIndex);
        //队列大小-1
        --count;
        //唤醒put线程
        notFull.signal();
        return x;
    }

  移除数据

  public boolean remove(Object o) {
        if (o == null) return false;
        final Object[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            //从takeIndex处开始计算,每次i加1,最大为队列最大容量count
            for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) {
                //如果移除元素在数组某个下标找到
                if (o.equals(items[i])) {
                    removeAt(i);
                    return true;
                }
            }
            return false;
        } finally {
            lock.unlock();
        }
    }
    void removeAt(int i) {
        final Object[] items = this.items;
        //如果准备移除的索引和下一个被取出的元素索引一样,直接移除
        if (i == takeIndex) {
            //赋值null,便于GC
            items[takeIndex] = null;
            //重新设置下一个被取出元素的索引
            takeIndex = inc(takeIndex);
        //如果需要删除的元素索引不是当前被取出的索引
        } else {
            //一直循环,直到删除为止
            for (;;) {
                //假设队列容量是4,目前存了3个元素,即takeIndex=0,putIndex=3,目前我打算删除数组下标为1的元素
                // nexti第一次为2
                int nexti = inc(i);
                if (nexti != putIndex) {
                    //相当于将队列往前移
                    items[i] = items[nexti];
                    //相当于i+1
                    i = nexti;
                //待删除的索引与待put的索引相等,比如putIndex=2,i=1,inc(i) = 2
                } else {
                    //索引i处置null,偏于GC
                    items[i] = null;
                    //重新赋值下一个即将放入元素的索引
                    putIndex = i;
                    break;
                }
            }
        }
        //队列大小-1
        --count;
        //唤醒put线程,公平的话按FIFO顺序,非公平的话可以抢占
        notFull.signal();
    }

    遍历队列

 public Iterator<E> iterator() {
        return new Itr();
    }
   
    private class Itr implements Iterator<E> {
        //队列里面还剩的元素个数
        private int remaining;
        //下一次调用next()返回的索引
        private int nextIndex;
        //下一次调用next()返回的元素
        private E nextItem; 
        //上一次调用next()返回的元素
        private E lastItem; 
        //上一次调用next()返回的索引 
        private int lastRet; 

        Itr() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                lastRet = -1;
                //只有队列里面还有元素
                if ((remaining = count) > 0)
                    //获取takeIndex处的元素
                    nextItem = itemAt(nextIndex = takeIndex);
            } finally {
                lock.unlock();
            }
        }

        public boolean hasNext() {
            return remaining > 0;
        }

        public E next() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                //如果队列没有值
                if (remaining <= 0)
                    throw new NoSuchElementException();
                lastRet = nextIndex;
                //获取下一次获取索引处的元素
                E x = itemAt(nextIndex);  // check for fresher value
                if (x == null) {
                    x = nextItem;         // we are forced to report old value
                    lastItem = null;      // but ensure remove fails
                }
                else
                 //将刚获取的元素当做上一次获取的元素
                    lastItem = x;
                 //当下一次获取的元素不存在的时候
                while (--remaining > 0 && // skip over nulls
                       (nextItem = itemAt(nextIndex = inc(nextIndex))) == null)
                    ;
                return x;
            } finally {
                lock.unlock();
            }
        }

        public void remove() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                int i = lastRet;
                if (i == -1)
                    throw new IllegalStateException();
                lastRet = -1;
                E x = lastItem;
                lastItem = null;
                // only remove if item still at index
                if (x != null && x == items[i]) {
                    boolean removingHead = (i == takeIndex);
                    removeAt(i);
                    if (!removingHead)
                        nextIndex = dec(nextIndex);
                }
            } finally {
                lock.unlock();
            }
        }
    }