ArrayBlockingQueue队列源码解析

秒速五厘米 2023-06-05 05:55 9阅读 0赞

#### ArrayBlockingQueue简介： ####

1.  ArrayBlockingQueue 是一个阻塞式的有界队列，内部存储数据的格式和 ArrayList 类似 通过 Object\[\] 数据来存储数据， 并且其在实现时通过 ReentrantLock 保证了线程的安全性。
2.  ArrayBlockingQueue 特点：  
    a. 初始化必须要定义队列的容量，且不会进行扩容( 容量满，则阻塞进队操作；容量空，则阻塞出队操作 )。  
    b. 先进先出队列。  
    c. 队列不保存为空的数据，入队时会进行验证： checkNotNull(e);

#### 常用方法： ####

1.  add(E e)：把 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则报异常
2.  offer(E e)：表示如果可能的话，将 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false
3.  put(E e)：把 e 加到 BlockingQueue 里，如果 BlockQueue 没有空间，则调用此方法的线程被阻断直到 BlockingQueue 里面有空间再继续
4.  poll(time)：取走 BlockingQueue 里排在首位的对象，若不能立即取出，则可以等 time 参数规定的时间,取不到时返回 null
5.  take()：取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 Blocking 有新的对象被加入为止
6.  remainingCapacity()：剩余可用的大小。等于初始容量减去当前的 size

##### 源码中参数介绍： #####

1.  其 takeIndex 和 putIndex 保证了 队列的先进先出功能
2.  lock 作为全局锁，保证数据安全
3.  源码：

// 存储数据
        final Object[] items;
    
        // 出队的位置
        int takeIndex;
    
        // 入队的存放位置
        int putIndex;
    
        // 队列的 元素个数
        int count;
    
        // 全局锁
        final ReentrantLock lock;
    
        // 出队时队列为空 Condition
        private final Condition notEmpty;
    
        // 入队时队列已经满 Condition
        private final Condition notFull;

##### 初始化方法： #####

1.  ArrayBlockingQueue 初始化有三种，通过不同的参数，新建不同情况的队列，参数：  
    a. int capacity: 队列的大小，指定后不可以修改，队列不会进行扩容  
    b. boolean fair : 指定对象中线程阻塞时是否是公平锁，由 ReentrantLock 来实现  
    c. Collection<? extends E> c : 其他集合中的数据，初始化到 ArrayBlockingQueue
2.  源码示例：

/**
         * ArrayBlockingQueue 初始化
         * 1. capacity 队列的大小，指定后不可以修改，队列不会进行扩容
         */
        public ArrayBlockingQueue(int capacity) {
            this(capacity, false);
        }
    
        /**
         * ArrayBlockingQueue 初始化
         * 1. capacity 队列的大小，指定后不可以修改，队列不会进行扩容
         * 2. fair 决定创建的 全局锁 是否为公平锁
         */
        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 初始化
         * 1. capacity 队列的大小，指定后不可以修改，队列不会进行扩容
         * 2. fair 决定创建的 全局锁 是否为公平锁
         * 3. c 初始化时将其他集合中的元素 保存到队列中
         * 4. 通过 lock加锁
         */
        public ArrayBlockingQueue(int capacity, boolean fair,
                                  Collection<? extends E> c) {
            this(capacity, fair);
    
            final ReentrantLock lock = this.lock;
            lock.lock(); // Lock only for visibility, not mutual exclusion
            try {
                int i = 0;
                try {
                    for (E e : c) {
                        checkNotNull(e);
                        items[i++] = e;
                    }
                } catch (ArrayIndexOutOfBoundsException ex) {
                    throw new IllegalArgumentException();
                }
                count = i;
                putIndex = (i == capacity) ? 0 : i;
            } finally {
                lock.unlock();
            }
        }

##### 入队方法： #####

1.  enqueue ：入队方法，私有的方法，供 ArrayBlockingQueue 内部调用，用于完成各种方式的入队， 提供给 add、offer、put 方法内部调用
2.  add(E e)：把 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则报异常
3.  offer(E e)：表示如果可能的话，将 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false  
    put(E e)：把 e 加到 BlockingQueue 里，如果 BlockQueue 没有空间，则调用此方法的线程被阻断直到 BlockingQueue 里面有空间再继续
4.  源码：

/**
         * 入队
         * 1. 根据 putIndex 把需要入队的 E 放在指定的位置
         * 2. 此处是 private 方法，内部调用，因此此处未加锁
         * 3. 队列不为空，通过notEmpty.signal()唤醒可能正在等待出队的线程
         */
        private void enqueue(E x) {
            // assert lock.getHoldCount() == 1;
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            notEmpty.signal();
        }
    
    	 /**
         * 1. 把 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则报异常
         * 2. 下面第二个 add 方法  是AbstractQueue.add()的方法，放到这里便于理解
         */
        public boolean add(E e) {
            return super.add(e);
        }
    
        // super.add(e) 中代码
        //public boolean add(E e) {
        //    if (offer(e))
        //        return true;
        //    else
        //        throw new IllegalStateException("Queue full");
        //}
    
        /**
         * 1. 表示如果可能的话，将 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false
         * 2. 上面的Add方法实际调用的是 offer方法, offer方法最终调用 enqueue入队
         * 3. 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();
            }
        }
    
    
        /**
         * 1. 表示如果可能的话，将 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false
         * 2. 上面的Add方法实际调用的是 offer方法, offer方法最终调用 enqueue入队
         * 3. offer方法中加锁保证线程安全
         * 4.
         */
        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();
            }
        }
    
        /**
         * 1. 把 e 加到 BlockingQueue 里，如果 BlockQueue 没有空间，则调用此方法的线程被阻断直到 BlockingQueue 里面有空间再继续
         * 2. put方法最终调用 enqueue入队
         * 3. put方法中加锁保证线程安全
         */
        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();
            }
        }

##### 出队方法： #####

1.  dequeue: 出队基础方法 和 入队 enqueue 相对应，都是 private 方法，提供给 poll、take 方法调用：
2.  poll(time)：取走 BlockingQueue 里排在首位的对象，若不能立即取出，则可以等 time 参数规定的时间,取不到时返回 null
3.  take()：取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 Blocking 有新的对象被加入为止
4.  源码：

/**
         * 出队
         * 1. 根据 takeIndex 从 items 取出需要出队的参数
         * 2. 此处是 private 方法，内部调用，因此此处未加锁
         * 3. 队列未满，最后通过 notFull.signal()唤醒可能正在等待入队的线程
         */
        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;
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;
            if (itrs != null)
                itrs.elementDequeued();
            notFull.signal();
            return x;
        }
    
        /**
         * 1. 无参方法
         * 2. 取走 BlockingQueue 里排在首位的对象
         * 3. 调用 dequeue 取出出队数据
         */
        public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : dequeue();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 1. 有参方法
         * 2. 取走 BlockingQueue 里排在首位的对象，若不能立即取出，则可以等 time 参数规定的时间,取不到时返回 null
         *  3. 调用 dequeue 取出出队数据
         */
        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;
                    nanos = notEmpty.awaitNanos(nanos);
                }
                return dequeue();
            } finally {
                lock.unlock();
            }
        }
        
        /**
         * 1. 取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 Blocking 有新的对象被加入为止
         * 2. 调用 dequeue 取出出队数据
         * 3. 已加锁
         */
        public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0)
                    notEmpty.await();
                return dequeue();
            } finally {
                lock.unlock();
            }
        }

##### 其他方法： #####

1.  size ： 获取队列的大小
2.  peek： 获取下一个出队的元素
3.  源码：

/**
         *  队列中元素的个数
         */
        public int size() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 获取下一个出队的元素
         * @return
         */
        public E peek() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return itemAt(takeIndex); // null when queue is empty
            } finally {
                lock.unlock();
            }
        }
        
        /**
         * 返回当前位置的数据，但不出队
         */
        final E itemAt(int i) {
            return (E) items[i];
        }

##### End ！ #####