JDK容器学习之Queue: ArrayBlockingQueue

基于数组阻塞队列 ArrayBlockingQueue

前面学习了基于数组的非阻塞双端队列ArrayDeque，其内部维护一个数组和指向队列头和队列尾索引的两个成员变量；本篇则探究下基于数组的阻塞队列是什么样的数据结构，又有什么特性，相较于ArrayDeque又有什么异同；而后就是使用场景了

I. 底层数据结构

先看内部成员变量定义， 和 ArrayDequeue相比，差异不大，一个数组，两个索引；此外多了一个锁和两个断定条件

/** 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;

/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;

注意

1. 底层数据结构依然是数组，注释上并无说明要求容量是2的n次方（ArrayDequeue有这个强制限定）
2. 初始化时必须指定队列的容量（即数组的长度，构造方法中的必选参数）
3. count直接表示队列的元素个数（注意DelayQueue是经过遍从来获取队列长度，且并发修改会有问题，那么这个是如何保证并发的？）
4. 留一个疑问，塞入队列超过数组容量，是否会出现扩容

数据结构以下图


II. 阻塞实现原理

0. Prefer

分析阻塞原理以前，先经过注释解释下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;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

阻塞方式的进队实现以下

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();
    }
}

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) {
        // 阻塞，知道队列不满
        // 或者超时时间已过，返回false
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}

源码分析，阻塞入队的逻辑比较清晰，小结一下

• 非阻塞调用方式 offer(e)
• 阻塞调用方式 put(e)或 offer(e, timeout, unit)
• 阻塞调用时，唤醒条件为超时或者队列非满（所以，要求在出队时，要发起一个唤醒操做）
• 进队成功以后，执行notEmpty.signal()唤起被阻塞的出队线程

2. 出队

非阻塞出队方法以下

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : 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;
}

阻塞的实现，逻辑比较清晰，首先竞争锁，判断是否为空，是阻塞直到非空；不然弹出队列头元素

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return 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);
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}

小结

• 非阻塞出队调用 poll()方法
• 阻塞出队调用 take()或 poll(long,TimeUnit)方法
• 出队以后，会唤醒因队列满被阻塞的入队线程
• 出队count计数减1（由于每次只能一个线程出队，因此count的值能够保证并发准确性）
• 支持在遍历队列时，出队

III. 应用场景及case

建立线程池时，一般会用到 ArrayBlockingQueue或者LinkedBlockingQueue，如

new ThreadPoolExecutor(1, 1,
          0L, TimeUnit.MILLISECONDS,
          new ArrayBlockingQueue<Runnable>(2));

延迟队列也是并发安全，ArrayBlockingQueue 相比较 DelayQueue应用场景的区别主要在

• 有界和无界（ArrayBlockingQueue不会扩容，而DelayQueue会出现扩容）
• 前者队列非空就能够出队；后者则须要队列头生效(getDelay()返回值小于0)
• 前者队列满，则没法入队；后者一直均可以入队
• 前者FIFO；后者优先级队列，延迟时间小的优先出队

IV. 小结

基于数组阻塞队列ArrayBlockingQueue

• 有界数组阻塞队列，FIFO，先进先出，初始化时指定数组长度
• 阻塞出队方法: take() 或 poll(long, TimeUnit)
• 非阻塞出队方法: poll()
• 阻塞入队方法: offer(E, long, TimeUnit)或put(E)
• 非阻塞入队方法: offer(E) add(E)
• 队列为空，出队会被阻塞；队列满时，进队会被阻塞
• 根据内部计数count，能够直接获取队列长度（count的并发安全是由进队出队上锁，保证同一时刻只有一个线程修改count值保证的）

更多java分享