ArrayBlockIngQueue源码原理用法解析

青旅半醒 2024-04-18 15:46 188阅读 0赞

ArrayBlockIngQueue的类结构

![70][]

如图所示，ArrayBlockingQueue内部有个数组items用来存放队列元素，putindex下标标示入队元素下标，takeindex是出对下标，count统计队列元素个数。从定义可知ArrayBlockingQueue没有使用volatile修饰，因为访问这些变量使用都是在锁块内，并不存在可见性问题。 另外有个独占锁lock用来对出入队操作加锁，这导致同时只有一个线程可以访问入队出队，另外notEmpty，notFull条件变量用来进行出入队的同步。

/** The queued items */
        final Object[] items;
    
        /** items index for next take, poll, peek or remove */
        int takeIndex;
    
        /** items index for next put, offer, or add */
        int putIndex;
    
        /** Number of elements in the queue */
        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;
    
        /** Condition for waiting puts */
        private final Condition notFull;

### 构造函数 ###

ArrayBlockingQueue的构造函数必须传入队列大小，所以是有界队列，默认是Lock，为非公平锁（false）。

/**
         * Creates an {@code ArrayBlockingQueue} with the given (fixed)
         * capacity and default access policy.
         *
         * @param capacity the capacity of this queue
         * @throws IllegalArgumentException if {@code capacity < 1}
         */
        public ArrayBlockingQueue(int capacity) {
            this(capacity, false);
        }
    
        /**
         * Creates an {@code ArrayBlockingQueue} with the given (fixed)
         * capacity and the specified access policy.
         *
         * @param capacity the capacity of this queue
         * @param fair if {@code true} then queue accesses for threads blocked
         *        on insertion or removal, are processed in FIFO order;
         *        if {@code false} the access order is unspecified.
         * @throws IllegalArgumentException if {@code capacity < 1}
         */
        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();
        }

公平锁：在并发环境中，每个线程在获取锁时会先查看此锁维护的等待队列。如果等待队列为空，或者当前线程是等待队列的第一个，就占有锁，否则当前线程就加入到等待队列中，然后按照FIFO的规则，从队列中取到自己。

非公平锁：不会查看锁维护的等待队列，而是直接尝试占有锁。如果占有失败，才遵循公平锁的规则。

### offer方法 ###

在队尾插入元素，如果队列满了，就返回false，否则元素入队，并返回true。

/**
         * Inserts the specified element at the tail of this queue if it is
         * possible to do so immediately without exceeding the queue's capacity,
         * returning {@code true} upon success and {@code false} if this queue
         * is full.  This method is generally preferable to method {@link #add},
         * which can fail to insert an element only by throwing an exception.
         *
         * @throws NullPointerException if the specified element is null
         */
        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();
            }
        }
    
        /**
         * Inserts element at current put position, advances, and signals.
         * Call only when holding lock.
         */
        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();
        }

①由于在操作共享变量前加了锁，所以不存在内存不可见问题，加过锁后获取的共享变量都是从主内存中获取的，而不是在CPU缓存或者寄存器里面的值，释放锁后修改的共享变量值会刷新回主内存中。

②队列是用循环数组实现的，所以计算下一个元素存放的下标在队满之后变为0，这点要注意。

③notEmpty.signal()是为了唤醒调用notEmpty.await()阻塞后放入notEmpty条件队列中的线程。下面是wait和await需要唤醒线程时候使用的方法对应关系

wait    notify/notifyAll  
await    signal/signalAll

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2Njg4MTQz_size_16_color_FFFFFF_t_70][]

### put方法 ###

在队列尾部添加元素，如果队列满了就等待，当前线程就会阻塞，直到出队操作调用了notFull.signal()方法唤醒该线程，然后该线程发现队列有空位置就插入，put方法没有返回值

/**
         * Inserts the specified element at the tail of this queue, waiting
         * for space to become available if the queue is full.
         *
         * @throws InterruptedException {@inheritDoc}
         * @throws NullPointerException {@inheritDoc}
         */
        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();
            }
        }

注意：这里为什么不是像offer方法一样加锁lock.lock()，而是使用lock.lockInterruptibly()？

因为调用了await()方法，await()方法在中断标志设置后会抛出InterruptedException异常，然后退出。所以put在需要加锁的时候，先判断中断标志是否设置，如果设置了就直接抛出InterruptedException异常，就不需要再去获取锁了。

### poll方法 ###

从队头获取并删除元素，队列如果为空，就返回null

public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return (count == 0) ? null : dequeue();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * Extracts element at current take position, advances, and signals.
         * Call only when holding lock.
         */
        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;
        }

注意这句：if ( ++takeIndex == items.length )  takeIndex = 0;

我们要时刻记住ArrayBlockingQueue内部是使用一个循环数组来存放元素的，所以即使你是删除元素（元素出队），也要判断当前出队下标takeIndex的下一个出队元素，对应下标是否队列尾部

### take方法 ###

从队头取元素，如果队列为空，当前线程阻塞，然后就会被挂起放到notEmpty的条件队列里面，直到入队操作使用notEmpty.signal()唤醒该线程，该线程发现队列有元素了才返回队头元素

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

### peek方法 ###

返回队列头元素，但不移除该元素，队列为空，返回null

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

### size方法 ###

获取队列元素个数，因为加了独占锁，其他线程不能入队或者出队、删除元素等操作，所以保证了线程安全

/**
         * Returns the number of elements in this queue.
         *
         * @return the number of elements in this queue
         */
        public int size() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return count;
            } finally {
                lock.unlock();
            }
        }

### 小结 ###

ArrayBlockingQueue通过使用全局独占锁实现了同时只能有一个线程进行入队或者出队操作，这个锁的粒度比较大，有点类似在方法上添加synchronized的意味。

其中offer，poll操作通过简单的加锁进行入队出队操作，而put，take则使用了条件变量实现如果队列满则等待，如果队列空则等待，然后分别在出队和入队操作中发送信号激活等待线程实现同步。

相比linkedBlockingQueue，ArrayBlockingQueue的size()方法的结果是准确的，因为计算前加了全局锁。

### ArrayBlockingQueue使用示例 ###

需求：在多线程操作下，一个数组最多只能存入3个元素。多放入不可以存入数组，或者等待某线程对数组中某个元素取走才能放入

思路：多放入不可以存入数组，或者等待某线程对数组中某个元素取走才能放入

“等待”：证明有线程间的唤醒、挂起状态，所以入队操作排除offer方法，我们应该使用put方法，出队操作排除peek方法。那么现在就是在poll方法和take方法之间，它们的区别就是在数组中没有值的时候是继续等待，还是返回空，我个人觉得都是可以的。但是题目中对于元素出队没有提到移除字样，那么就认为不需要移除，所以出队操作我们选择take方法

public class BlockingQueueTest {
        public static void main(String[] args) {
            final BlockingQueue queue = new ArrayBlockingQueue(3);
            // 开了两个线程
            for (int i = 0; i < 2; i++) {
                new Thread(()->{
                    while (true){
                        try {
                            Thread.sleep((long)Math.random()*1000);
                            System.out.println(Thread.currentThread().getName() + " 循环里的线程 准备放数据");
    
                            queue.put(1);
                            System.out.println(Thread.currentThread().getName() + " 循环里的线程 已经放了数据，目前数据个数：" + queue.size());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
            }
    
            new Thread(()->{
                while (true){
                    try {
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + " 另一个循环的线程 准备取数据");
    
                        System.err.println(queue.take());
                        System.out.println(Thread.currentThread().getName() + " 另一个循环的线程 已经取走数据，目前数据个数：" + queue.size());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
    
        }
    
    }

注意：元素出队操作的提示我选择了System.err.println，为什么使用err而不使用out，可以点击[System.out和System.err分析][System.out_System.err]

另外，从[https://blog.csdn.net/li\_canhui/article/details/84100166][https_blog.csdn.net_li_canhui_article_details_84100166]我看到这个作者的两个疑问，挺好的，就搬运过来：

1、如果当前队列慢了，执行put的线程在获取到锁之后，等待notFull条件上。那么，当执行take操作的线程想获取锁时，阻塞队列的锁已经被前面put的线程获取了，那么take将永远得不到机会执行。怎么回事呢？

后来，我查了condition的await方法，它的注释如下：

Causes the current thread to wait until it is signalled or interrupted.

The lock associated with this Condition is atomically released and the current thread becomes disabled for thread scheduling purposes and lies dormant until one of four things happens......

原因在await方法的作用上。因为condition是通过lock创建的，而调用condition的await方法时，会自动释放和condition关联的锁。所以说，当put线程被阻塞后，它实际已经释放了锁了。所以，当有take线程想执行时，它是可以获取到锁的。

2、当等待在condition上的线程被唤醒时，因为之前调用await前，已经获取了锁，那么被唤醒时，它是自动就拥有了锁，还是需要重新获取呢？

在await方法的注释中，有如下的一段话：

In all cases, before this method can return the current thread must re-acquire the lock associated with this condition. When the thread returns it is guaranteed to hold this lock.

说明当等待在condition上的线程被唤醒时，它需要重新获取condition关联的锁，获取到之后，await方法才会返回。

[70]: https://img-blog.csdn.net/20180507180620357?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FuZGFtYWppbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2Njg4MTQz_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/6db257db1c6841f89353b9eb5a7405a0.png
[System.out_System.err]: https://blog.csdn.net/qq_36688143/article/details/100666015
[https_blog.csdn.net_li_canhui_article_details_84100166]: https://blog.csdn.net/li_canhui/article/details/84100166