Java8集合系列之ArrayBlockingQueue（九）

Posted on   |   In   |     |   visitors

上一篇是LinkedBlockingQueue，当中阐述了阻塞队列和非阻塞队列的区别。ArrayBlockingQueue也是一种阻塞队列，底层是以数组为数据结构，而且在初始化的时候必须指定容量上限。

继承树

继承关系如下，与LinkedBlockingQueue一样：

1
2
ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable

主要属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/** 队列元素数组 */
final Object[] items;
/** 对于下一次take/poll/peek/ remove操作的元素索引,默认为0 */
int takeIndex;
/** 对于下一次put/offer/add 操作的元素索引，默认为0*/
int putIndex;
/**队列中元素数量*/
int count;
///////////以下变量用于控制并发操作//////////////
/**唯一一把锁控制所有入口*/
final ReentrantLock lock;
/** 用于等待take操作的条件 */
private final Condition notEmpty;
/** 用于等待put操作的条件*/
private final Condition notFull;

主要方法

构造方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
/**
 *如果fair为false，则所有的阻塞线程在插入和移除元素的顺序都按照先进先出的顺序（FIFO）
 */
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();
}
//索性看看如何创建两种不同公平策略的锁，代码如下;
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

offer/poll/peek

关于add/remove/element方法的实现，已经在LinkedBlockingQueue写过，都是在AbstractQueue实现而成，只是对offer/poll/peek的包装，所以在此只分析后者的实现策略。
offer方法代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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();
    }
}
//入队操作
private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    items[putIndex] = x;//1.先赋值
    if (++putIndex == items.length)//2.再加1
        putIndex = 0;//到达末尾又从0开始
    count++;
    notEmpty.signal();
}

poll代码更简单，如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}
//出队操作
private E dequeue() {
    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;
}

至于peek方法只读取队首元素而不弹出，实现更加简单，在此不做赘述。

put/take

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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 E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

以上方法更见常用，还有以延迟等待若干时间思想实现的offer/poll方法亦是非常简单不多做详述。则以数组实现的阻塞队列大抵如此，总结几点如下：

  1. 首部和尾部数组下标索引在数组中顺时针循环，从小到大。
  2. 添加元素，先赋值，索引加1；弹出元素亦如是，先弹出元素，再加1。
  3. 索引到达数组上限边界，则继续从0开始。
  4. 不论队首、队尾，所有并发操作共用一把锁lock，效率更较。

在这个集合系列末尾，提一下Doug Lea这个大神，java.util.concurrent（J.U.C）中的绝大部分源码的作者，感兴趣的可以去了解，几乎是J.U.C的缔造者，向大神致敬。

-EOF-