java阻塞队列之ArrayBlockingQueue

ArrayBlockingQueue简介

它是 BlockingQueue 接口的有界队列实现类，底层采用数组来实现。

ArrayBlockingQueue的特性

1.其并发控制采用可重入锁来控制，不管是插入操作还是读取操作，都需要获取到锁才能进行操作，它采用一个 ReentrantLock 和相应的两个Condition来实现。

2.采用FIFO的原则对元素进行排序添加

3.既然是有界，它在示例化时指定了数组容量

ArrayBlockingQueue的重要属性

// 用于存放元素的数组 final Object[] items；

// 下一次读取操作的位置 int takeIndex；

// 下一次写入操作的位置 int putIndex；

// 队列中的元素数量 int count；

// 以下几个就是控制并发用的同步器

final ReentrantLock lock；

private final Condition notEmpty；

private final Condition notFull；

对于 ArrayBlockingQueue，我们可以在构造的时候指定以下三个参数：

1.队列容量，其限制了队列中最多允许的元素个数；

2.指定独占锁是公平锁还是非公平锁。非公平锁的吞吐量比较高，公平锁可以保证每次都是等待最久的线程获取到锁；

3.可以指定用一个集合来初始化，将此集合中的元素在构造方法期间就先添加到队列中。

ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)
ArrayBlockingQueue(int capacity, boolean fair)
           

ArrayBlockingQueue的同步机制

ArrayBlockingQueue实现并发同步的原理就是，读操作和写操作都需要获取到 AQS 独占锁才能进行操作。如果队列为空，这个时候读操作的线程进入到读线程队列排队，等待写线程写入新的元素，然后唤醒读线程队列的第一个等待线程。如果队列已满，这个时候写操作的线程进入到写线程队列排队，等待读线程将队列元素移除腾出空间，然后唤醒写线程队列的第一个等待线程。

入队方法

该方法有两个重载方法offer(E e，long timeout，TimeUnit unit)和offer(E e)。 将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量），在成功时返回 true，如果此队列已满，则返回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();
        }
    }

private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0; // 队列满了，重置putIndex为0
        // 这是ArrayBlockingQueue最重要的特性之一：队列实现中的数组会被循环利用
        // 当放满之后，前面的位置如果空下来了，又会将putIndex指向0，在前面的位置依次放数据。
    count++;
    notEmpty.signal();
}
           

put(E e) 

将指定的元素插入到队列的尾部，如果有可用空间直接插入，如果没有可用空间，调用condition.await()方法等待，直到被唤醒，然后插入元素。

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();//如果没有可用空间，调用await方法阻塞线程
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }
           

add(E e)

如果插入元素成功，返回true，如果插入失败抛出异常IllegalStateException(“Queue full”)。

public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
           

出队方法

该方法也有两个重载方法poll(long timeout, TimeUnit unit)和poll()，从队列头部移除一个元素，前者与后者的区别在于，如果队列中没有可以移除的元素，前者会等待一定时间，然后执行移除方法。

public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : dequeue();
            //如果没有可以移出元素，返回null,否则执行dequeue()方法
        } finally {
            lock.unlock();
        }
    }
    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);
                //如果没有可以移出元素，调用condition的线程等待的方法，等待一定时间
            }
            return dequeue();
        } finally {
            lock.unlock();//最后释放锁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;
    }
           

remove(Object o) 

移除指定的元素，如果不存在返回false。

take() 

获取并移除此队列的头部。take()和poll()的区别在于，如果队列中没有可移除元素，take()会一直等待，而poll()可设置直接返回null或者等待一定时间。

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();//如果队列中没有元素，该线程一直处于阻塞状态
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
           

参考：https://blog.csdn.net/qq_27770257/article/details/79622453