ArrayBlockingQueue 源码分析

JavaSE

TOP 带着问题看源码

  1. ArrayBlockingQueue 实现原理
  2. ArrayBlockingQueue 应用的场景
  3. ArrayBlockingQueue 和 LinkedBlockingQueue 区别

1. 基本介绍

ArrayBlockingQueue 是一个基于数组实现的线程安全的阻塞队列。

其实现了阻塞队列 BlockingQueue 接口和基本队列操作 AbstractQueue 接口

img

2. 成员变量分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 存放数据数组
final Object[] items;
// 下一个取的索引
int takeIndex;
// 下一个放的索引
int putIndex;
// 队列数量
int count;
// 锁
final ReentrantLock lock;
// 队列不为空条件
private final Condition notEmpty;
// 队列没有满条件
private final Condition notFull;
// 迭代器状态
transient Itrs itrs = null;

3. 构造方法分析

1
2
3
4
5
6
7
8
9
10
11
12
13
public ArrayBlockingQueue(int capacity) {
  	// 默认非公平锁，调用下个构造方法
    this(capacity, false);
}
// 初始化数组，锁，条件变量
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();
}

4. 核心方法分析

4.1 入队列操作

4.1.1 enqueue(E x)

核心逻辑就是往数组中插入一条数据，然后更新 putIndex，唤醒 “队列不为空” 条件对应的条件队列。

1
2
3
4
5
6
7
8
9
10
11
12
private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    items[putIndex] = x;
  	// 因为取也是从0开始取，所以接下来已经满了就重置为0
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
  	// 因为已经有数据了，就可以唤醒 notEmpty Condition了
    notEmpty.signal();
}

4.1.2 offer(E e)

队列对外暴露的入队列 API，其实就是调用了一下 enqueue 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public boolean offer(E e) {
  	// 如果传的是null 直接抛空指针异常
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
      	// 如果已经满了，返回false
        if (count == items.length)
            return false;
        else {
          	// 调用 enqueue 插入数据
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

4.1.3 put(E e)

队列对外暴露的入队列 阻塞API，如果队列已满，会进入阻塞状态，直到 ”队列没有满“ 条件满足，才会插入新的数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
  	// 可响应中断
    lock.lockInterruptibly();
    try {
      	// 如果队列已满
        while (count == items.length)
          	// 等待 ”队列没有满“ Condition
            notFull.await();
      	// 插入数据
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

4.2 出队列操作

4.2.1 dequeue()

核心逻辑就是从数组中取出一条数据，然后更新 takeIndex，唤醒 “队列没有满” 条件对应的条件队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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;
  	// 如果取完就重置 takeIndex，继续从0开始取
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
  	// 走到这里说明已经取一条了
  	// 这个时候需要唤醒 "队列没有满" Condition
    notFull.signal();
    return x;
}

4.2.2 poll()

队列对外暴露的取队列 API，如果取完，返回null，没取完就调用 dequeue 方法

1
2
3
4
5
6
7
8
9
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}

4.2.3 take()

队列对外暴露的取队列 阻塞API，如果队列已空，会进入阻塞状态，直到 “队列不为空” 条件满足，才会继续取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
  	// 可响应中断
    lock.lockInterruptibly();
    try {
      	// 队列已空
        while (count == 0)
          	// 等待 ”队列不为空“ Condition
            notEmpty.await();
      	// 调用 dequeue 取元素
        return dequeue();
    } finally {
        lock.unlock();
    }
}

4.2.4 peek()

队列对外暴露的只看不取 API，逻辑很简单，就是直接按照 takeIndex 查看元素，但是不置空也不维护 takeIndex

1
2
3
4
5
6
7
8
9
10
11
12
13
public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return itemAt(takeIndex); // null when queue is empty
    } finally {
        lock.unlock();
    }
}

final E itemAt(int i) {
    return (E) items[i];
}

通过上面的核心方法分析，回到问题 TOP 1 我们可以明白其实现原理就是采用 可重入非公平锁来保证线程安全，通过在数组的入口和出口来互相更新条件变量的唤醒条件来实现阻塞队列。

5. 总结

整体设计我们可以发现设计的很巧妙，既有阻塞 API，也有不阻塞线程安全的 API。回到问题 TOP 2 其应用场景不难想象，只要是涉及到内存中的生产者-消费者模型的都可以使用它来暂存数据。

关于最后的问题 TOP 3 ，其区别主要是底层数据结构的选择，整体的阻塞设计和逻辑都是一样的。