Java集合--ArrayBlockingQueue

超、凢脫俗 2022-06-12 10:40 121阅读 0赞

ArrayBlockingQueue是数组实现的线程安全的有界的阻塞队列。  
线程安全是指，ArrayBlockingQueue内部通过“互斥锁”保护竞争资源，实现了多线程对竞争资源的互斥访问。而有界，则是指ArrayBlockingQueue对应的数组是有界限的。 阻塞队列，是指多线程访问竞争资源时，当竞争资源已被某线程获取时，其它要获取该资源的线程需要阻塞等待；而且，ArrayBlockingQueue是按 FIFO（先进先出）原则对元素进行排序，元素都是从尾部插入到队列，从头部开始返回。

注意：ArrayBlockingQueue不同于ConcurrentLinkedQueue，ArrayBlockingQueue是数组实现的，并且是有界限的；而ConcurrentLinkedQueue是链表实现的，是无界限的。

ArrayBlockingQueue的数据结构，如下图所示：

[![30232555-bceb5a739eff444da550ad5c0994b341.jpg][]][30232555-bceb5a739eff444da550ad5c0994b341.jpg]

**说明**：  
    1. ArrayBlockingQueue继承于AbstractQueue，并且它实现了BlockingQueue接口。  
    2. ArrayBlockingQueue内部是通过Object\[\]数组保存数据的，也就是说ArrayBlockingQueue本质上是通过数组实现的。ArrayBlockingQueue的大小，即数组的容量是创建ArrayBlockingQueue时指定的。  
    3. ArrayBlockingQueue与ReentrantLock是组合关系，ArrayBlockingQueue中包含一个ReentrantLock对象(lock)。ReentrantLock是可重入的互斥锁，ArrayBlockingQueue就是根据该互斥锁实现“多线程对竞争资源的互斥访问”。而且，ReentrantLock分为公平锁和非公平锁，关于具体使用公平锁还是非公平锁，在创建ArrayBlockingQueue时可以指定；而且，ArrayBlockingQueue默认会使用非公平锁。  
    4. ArrayBlockingQueue与Condition是组合关系，ArrayBlockingQueue中包含两个Condition对象(notEmpty和notFull)。而且，Condition又依赖于ArrayBlockingQueue而存在，通过Condition可以实现对ArrayBlockingQueue的更精确的访问 -- (01)若某线程(线程A)要取数据时，数组正好为空，则该线程会执行notEmpty.await()进行等待；当其它某个线程(线程B)向数组中插入了数据之后，会调用notEmpty.signal()唤醒“notEmpty上的等待线程”。此时，线程A会被唤醒从而得以继续运行。(02)若某线程(线程H)要插入数据时，数组已满，则该线程会它执行notFull.await()进行等待；当其它某个线程(线程I)取出数据之后，会调用notFull.signal()唤醒“notFull上的等待线程”。此时，线程H就会被唤醒从而得以继续运行。

ArrayBlockingQueue提供的数据存储数组：

/** The queued items */
        final Object[] items;

ArrayBlockingQueue提供了一个互斥锁lock用来保证多线程对竞争资源的互斥访问。

提供两个条件变量notEmpty和notFull用来标识items数组数据的空还是满，

/** 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的添加元素和获取元素操作。

1、添加

简单来说就是获取锁之后尝试添加，如果已满之间返回false，否则在数组中添加数据，最后释放锁。

public boolean offer(E e) {
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
    		//获取锁
            lock.lock();
            try {
                if (count == items.length)
    				//队列已满返回false
                    return false;
                else {
    				//在队列中添加数据
                    enqueue(e);
                    return true;
                }
            } finally {
    			//释放锁
                lock.unlock();
            }
        }

private void enqueue(E x) {
            // assert lock.getHoldCount() == 1;
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            items[putIndex] = x;
    		//没添加一个元素putindex++
            if (++putIndex == items.length)
                putIndex = 0;
    		//数量加一
            count++;
    		//唤醒阻塞线程
            notEmpty.signal();
        }

2、获取数据

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;
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;
            if (itrs != null)
                itrs.elementDequeued();
            notFull.signal();
            return x;
        }

总结：ArrayBlockingQueue提供了一个有长度的数组items用来保存数据，并且实现队列的FIFO；提供互斥锁lock来保存竞争资源的互斥方法；提供两个信号量notEmpty和notFull来通知items空和满的状态。

[30232555-bceb5a739eff444da550ad5c0994b341.jpg]: /images/20220612/f7d99afef8a34758a054e0a55da7485c.png