美文网首页
ArrayBlockingQueue源码分析

ArrayBlockingQueue源码分析

作者: A_Xin | 来源:发表于2018-08-08 11:53 被阅读0次

    ArrayBlockingQueue源码分析

    ArrayBlockingQueue是一个用数组实现的队列,所以在效率上比链表结构的LinkedBlockingQueue要快一些,但是队列长度固定,不能扩展,入列和出列使用同一把锁。LinkedBlockingQueue是入列出列两把锁,读写分离。

    image.png

    成员变量和构造方法

    1、成员变量

    //队列存放的数组
    final Object[] items;
    
    //下次take/remove/peek/poll的索引
    int takeIndex;
    
    //下次add/offer/put的索引
    int putIndex;
    
    //当前队列中元素的数量
    int count;
    
    //锁
    final ReentrantLock lock;
    
    //队列为空阻塞的Condition
    private final Condition notEmpty;
    
    //队列塞满阻塞的Condition
    private final Condition notFull;
    
    transient Itrs itrs = null;
    

    2、构造方法

    //传入参数为队列的容量,传入后队列容量固定,不能扩容
     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);
            //初始化用于阻塞的Condition
            notEmpty = lock.newCondition();
            notFull =  lock.newCondition();
        }
    

    从ArrayBlockingQueue的成员变量和构造方法可以看出,ArrayBlockingQueue是一个用数组实现的,容量固定的队列,

    主要方法

    1、offer(E e)

    往队列中添加一条元素,如果添加成功,返回true,添加失败则返回false

    public boolean offer(E e) {
            //元素校验
            checkNotNull(e);
            //引用锁
            final ReentrantLock lock = this.lock;
            //上锁
            lock.lock();
            try {
                //当队列塞满后,不能再继续往队列中添加元素,返回false
                if (count == items.length)
                    return false;
                else {
                //队列还未塞满,执行入列方法,入列成功返回true
                    enqueue(e);
                    return true;
                }
            } finally {
                //释放锁
                lock.unlock();
            }
        }
    

    这里有一点注意的是在LinkedBlockingQueue中入列之后有一个自我唤醒的方法,而这里却没有,是因为LinkedBlockingQueue的入列和出列是分别不同的两把锁,读写分离。而这里读写用的是同一把锁,所以在读和写在同一时间内只能执行一个方法,就不会存在线程假死状态。

    2、enqueue(E x)

    入列方法

    private void enqueue(E x) {
            final Object[] items = this.items;
            //入列:往数组中存入元素
            items[putIndex] = x;
            /**判断元素是否存到了数组的最后一个位置上,如果是,
            就把下一个元素入列的索引置为1,防止索引越界,*/
            if (++putIndex == items.length)
                putIndex = 0;
            //入列成功,当前队列元素数量自增
            count++;
            /**通知还在等待的出列方法,队列中已有元素,
            可以进行出列了*/
            notEmpty.signal();
        }
    

    3、offer(E e, long timeout, TimeUnit unit)

    入列,给定一个超时时间,如果队列塞满了,则进行超时等待(线程阻塞),超时后入列失败,返回false

    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)
                        return false;
                    nanos = notFull.awaitNanos(nanos);
                }
                //入列
                enqueue(e);
                return true;
            } finally {
                //锁释放
                lock.unlock();
            }
        }
    

    与其他offer超时等待方法一样,在队列塞满的时候都会进行超时等待,如果等待超时则入列失败,并返回false

    4、put(E e)

    往队列中添加一条元素,如果队列塞满了,则线程无限期等待。直到有出列方法执行后队列还有剩余空间,在出列方法中唤醒当前正在阻塞的入列线程,继续执行入列操作

    public void put(E e) throws InterruptedException {
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                //队列塞满了
                while (count == items.length)
                    //线程进行无限期等待
                    notFull.await();
                //执行入列方法
                enqueue(e);
            } finally {
                lock.unlock();
            }
        }
    

    5、add(E e)

    往队列中添加一条元素,如果添加失败,则抛出异常

    //调用父类的add方法
    public boolean add(E e) {
        return super.add(e);
    }
    
    //调用入列方法offer,offer入列失败,抛出异常,成功则返回true
    public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
    

    6、poll()

    从队列中取出一条元素,并删除,如果取出成功,返回被取出的元素,如果取出失败,则返回null

       public E poll() {
            final ReentrantLock lock = this.lock;
            //上锁
            lock.lock();
            try {
                //队列为空,返回null,否则执行出列方法
                return (count == 0) ? null : dequeue();
            } finally {
                //释放锁
                lock.unlock();
            }
        }
    

    7、poll(long timeout, TimeUnit unit)

    从队列中取出一条元素,并给定一个超时时间。如果队列为空,则进行超时等待。如果等待超时后返回null

    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            //超时时间转换成纳秒
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            //上锁,超时等待过程如果线程中断,则抛出异常
            lock.lockInterruptibly();
            try {
                //队列为空,超时等待,等待超时,返回null
                while (count == 0) {
                    if (nanos <= 0)
                        return null;
                    nanos = notEmpty.awaitNanos(nanos);
                }
                //执行出列方法
                return dequeue();
            } finally {
                //释放锁
                lock.unlock();
            }
        }
    

    8、E peek()

    从队列中取出第一条元素,但不移除

    public E peek() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return itemAt(takeIndex); // null when queue is empty
            } finally {
                lock.unlock();
            }
        }
    
    

    9、take()

    从队列中取出一条元素,如果队列为空,则线程进行无限期等待,直到有执行入列操作的线程入列成功,队列中有元素后,在入列方法中环信当前正在等待出列的线程进行出列操作

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

    10、dequeue()

    dequeue()出列方法,执行队列出列,并标记下一个元素出列的索引

    private E dequeue() {
            // 引用队列存放的数组
            final Object[] items = this.items;
            @SuppressWarnings("unchecked")
            //获取要出列的元素
            E x = (E) items[takeIndex];
            //移除出列后的元素
            items[takeIndex] = null;
            /**标记下次出列的元素的索引,并判断当前出列元素
            是否是数组中最后一条元素,如果是则,标记下次出列元素索引为0,从数组头部开始出列*/
            if (++takeIndex == items.length)
                takeIndex = 0;
            //出列成功,队列长度-1
            count--;
            if (itrs != null)
                itrs.elementDequeued();
            notFull.signal();
            return x;
        }
    

    11、remove()

    从队列中移除一条元素,如果移除成功,返回被移除的元素。如果移除失败,抛出异常

    public E remove() {
            E x = poll();
            if (x != null)
                return x;
            else
                throw new NoSuchElementException();
        }
    
    

    12、remove(Object o)

    移除指定元素。遍历队列,找到要移除的元素,并从队列移除

    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;
                    /**从出列位置开始循环查找数组中的元素,
                    直到找到了要删除的元素的索引,执行
                    removeAt(i)根据元素索引删除元素的方法,
                    最后返回true表示删除成功,否则返回false删除失败*/
                    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) {
            //引用数组
            final Object[] items = this.items;
            //如果要删除的元素就是队列中的第一个元素,直接移除,并且将出列索引+1,队列长度-1
            if (removeIndex == takeIndex) {
                // 
                items[takeIndex] = null;
                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;
                    if (next == items.length)
                        next = 0;
                    //中间的元素
                    if (next != putIndex) {
                        items[i] = items[next];
                        i = next;
                    //遍历到了最后一个元素
                    } else {
                        items[i] = null;
                        //下次入列的索引重置,并跳出循环
                        this.putIndex = i;
                        break;
                    }
                }
                count--;
                if (itrs != null)
                    itrs.removedAt(removeIndex);
            }
            notFull.signal();
        }
    

    其他方法

    1、element()

    获取队列中的第一条元素,如果获取成功则返回取出的元素,否则抛出异常

    2、size()

    获取队列当长度

    3、clear()

    清空队列

    4、drainTo(Collection<? super E> c)

    将队列转换成实现了Collection接口的数组

    5、drainTo(Collection<? super E> c, int maxElements)

    将队列的指定长度的元素转换成实现了Collection接口的数组,如果指定长度超过了队列长度,按队列长度为准

    6、remainingCapacity()

    获取队列的剩余容量

    7、contains()

    队列中是否包含某个元素

    相关文章

      网友评论

          本文标题:ArrayBlockingQueue源码分析

          本文链接:https://www.haomeiwen.com/subject/gnbtbftx.html