boolean add(E e);

boolean offer(E e);

void put(E e) throws InterruptedException;

E take() throws InterruptedException;

E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

boolean remove(Object o);

//存放元素
final Object[] items; 

//取元素的索引
int takeIndex;

//存元素的索引
int putIndex;

//元素的数量
int count;

//控制并发的锁
final ReentrantLock lock;

//非空条件信号量
private final Condition notEmpty;

//非满条件信号量
private final Condition notFull;

transient Itrs itrs = null;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

//ArrayBlockingQueue.put()
public void put(E e) throws InterruptedException {
    //非空检查
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    //加锁
    lock.lockInterruptibly();
    try {
        //数组满了，线程加入 notFull 队列中等待被唤醒
        while (count == items.length)
            notFull.await();
        //添加元素
        enqueue(e);
    } finally {
        //解锁
        lock.unlock();
    }
}

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
27

//ArrayBlockingQueue.poll()
public E poll() {
    final ReentrantLock lock = this.lock;
    //加锁
    lock.lock();
    try {
        //没有元素返回 null，否则元素出队
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}

//ArrayBlockingQueue.dequeue()
private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    //获取 takeIndex 上的元素
    E x = (E) items[takeIndex];
    //设置 takeIndex 索引上的元素为 null
    items[takeIndex] = null;
    //当 takeIndex 长度是数组长度，takeIndex 索引从 0 开始
    if (++takeIndex == items.length)
        takeIndex = 0;
    //元素数量 -1
    count--;

    if (itrs != null)
        //更新迭代器
        itrs.elementDequeued();
    //唤醒 notFull 的等待队列，其中等待的第一个线程可以添加元素了
    notFull.signal();
    return x;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

//ArrayBlockingQueue.take()
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    //加锁
    lock.lockInterruptibly();
    try {
        //无元素
        while (count == 0)
            //将线程加入 notEmpty 的等待队列中，等待被入队的元素唤醒
            notEmpty.await();
        //元素出队
        return dequeue();
    } finally {
        //解锁
        lock.unlock();
    }
}

//ArrayBlockingQueue.remove()
public boolean remove(Object o) {
    //非空检查
    if (o == null) return false;
    final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    //加锁
    lock.lock();
    try {

        if (count > 0) {
            //入队元素的索引
            final int putIndex = this.putIndex;
            //出队元素的索引
            int i = takeIndex;
            do {
                //找到元素
                if (o.equals(items[i])) {
                    removeAt(i);
                    return true;
                }
                //i 等于数组长度的时候，从 0 开始
                if (++i == items.length)
                    i = 0;
            // i == putIndex 说明已经遍历了一遍
            } while (i != putIndex);
        }
        return false;
    } finally {
        //解锁
        lock.unlock();
    }
}

//ArrayBlockingQueue.removeAt()
void removeAt(final int removeIndex) {
    final Object[] items = this.items;
    //需要出队的 removeIndex 正好是 takeIndex
    if (removeIndex == takeIndex) {
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        //更新迭代器
        if (itrs != null)
            itrs.elementDequeued();
    } else {
        final int putIndex = this.putIndex;
        // 循环移动元素，将 next 元素向前移动 1 个
        for (int i = removeIndex;;) {
            int next = i + 1;
            if (next == items.length)
                next = 0;
            if (next != putIndex) {
                items[i] = items[next];
                i = next;
            } else {
                //设置 i 索引的位置为空，putIndex 索引为 i
                items[i] = null;
                this.putIndex = i;
                break;
            }
        }
        count--;
        if (itrs != null)
            itrs.removedAt(removeIndex);
    }
    // 唤醒 notFull 队列中等待的线程，通知可以元素入队了
    notFull.signal();
}