ArrayBlockingQueue

// 队列元素
final Object[] items;

// 读取坐标
int takeIndex;

// 写入坐标
int putIndex;

// 当前数据个数
int count;

public ArrayBlockingQueue(int capacity, boolean fair) {
  if (capacity <= 0)
    throw new IllegalArgumentException();

  this.items = new Object[capacity];

  // 基于 Lock 的 await 和 signal 通知机制
  lock = new ReentrantLock(fair);

  // 实例化了两个 Condition 队列，用于区分 获取 和 添加
  // 唤醒的时候可以唤醒 指定类型的操作
  notEmpty = lock.newCondition();
  notFull =  lock.newCondition();
}

public void put(E e) throws InterruptedException {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  // 如果当前线程未被中断，则获取锁
  lock.lockInterruptibly();
  try {
    // 队列已经满了
    while (count == items.length)
      // park 阻塞
      notFull.await();
    
    // 入队
    enqueue(e);
  } finally {
    lock.unlock();
  }
}

private void enqueue(E x) {
  final Object[] items = this.items;
  items[putIndex] = x;
  // 如果下标已经移动到最后一位，从头开始。 
  // 相当于下标循环移动，只要 takeIndex 不超过 putIndex，就不会出现同一个数据读两次的情况
  // putIndex 也不会超过 takeIndex，因为入队之前做了 count 队里数据个数的判断
  if (++putIndex == items.length)
    putIndex = 0;
  
  // 统计队列元素个数
  count++;

  // 如果有数据了，唤醒因为 take 没有数据 阻塞的线程
  notEmpty.signal();
}

public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
  // 如果当前线程未被中断，则获取锁
  lock.lockInterruptibly();
  try {
    // 队列空了
    while (count == 0)
      // park 阻塞
      notEmpty.await();
    
    // 出队
    return dequeue();
  } finally {
    lock.unlock();
  }
}

private E dequeue() {
  final Object[] items = this.items;
  E x = (E) items[takeIndex];
  // 当前坐标数据制空
  items[takeIndex] = null;
  // 如果下标已经移动到最后一位，从头开始。 
  // 相当于下标循环移动，只要 takeIndex 不超过 putIndex，就不会出现同一个数据读两次的情况
  // putIndex 也不会超过 takeIndex，因为入队之前做了 count 队里数据个数的判断
  if (++takeIndex == items.length)
    takeIndex = 0;
	// 统计队列元素个数
  count--;
  
  // 迭代器操作
  if (itrs != null)
    itrs.elementDequeued();
  
  // 如果数据减少了，唤醒因为队列满了 put 阻塞的线程
  notFull.signal();
  return x;
}