JDK数组阻塞队列

    /** 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; // 统计队列当中一共有多少个数据
 
    /*
     * Concurrency control uses the classic two-condition algorithm
     * found in any textbook.
     */
 
    /** Main lock guarding all access */
    final ReentrantLock lock; // 因为阻塞队列是一种可以并发使用的数据结构
 
    /** Condition for waiting takes */
    private final Condition notEmpty; // 这个条件变量主要用于唤醒被 take 函数阻塞的线程 也就是从队列当中取数据的线程
 
    /** Condition for waiting puts */
    private final Condition notFull; // 这个条件变量主要用于唤醒被 put 函数阻塞的线程 也就是从队列当中放数据的线程

public ArrayBlockingQueue(int capacity) {
  // capacity 表示用与存储数据的数组的长度
  this(capacity, false);
}
// fair 这个参数主要是用于说明 是否使用公平锁
// 如果为 true 表示使用公平锁 执行效率低 但是各个线程进入临界区的顺序是先来后到的顺序 更加公平
// 如果为 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();
}

public void put(E e) throws InterruptedException {
  checkNotNull(e); // 保证输入的数据不为 null 代码在下方
  final ReentrantLock lock = this.lock;
  // 进行加锁操作，因为下面是临界区
  lock.lockInterruptibly();
  try {
    while (count == items.length) // 如果队列已经满了 也就是队列当中数据的个数 count == 数组的长度的话 就需要将线程挂起
      notFull.await();
    // 当队列当中有空间的之后将数据加入到队列当中 这个函数在下面仔细分析 代码在下方
    enqueue(e);
  } finally {
    lock.unlock();
  }
}
 
private static void checkNotNull(Object v) {
  if (v == null)
    throw new NullPointerException();
}
 
private void enqueue(E x) {
  // assert lock.getHoldCount() == 1;
  // assert items[putIndex] == null;
  // 进入这个函数的线程已经在 put 函数当中加上锁了 因此这里不需要加锁
  final Object[] items = this.items;
  items[putIndex] = x;
  if (++putIndex == items.length) // 因为这个数据是循环使用的 因此可以回到下标为0的位置
    // 因为队列当中的数据可以出队 因此下标为 0 的位置不存在数据可以使用
    putIndex = 0;
  count++;
  // 在这里需要将一个被 take 函数阻塞的线程唤醒 如果调用这个方法的时候没有线程阻塞
  // 那么调用这个方法相当于没有调用 如果有线程阻塞那么将会唤醒一个线程
  notEmpty.signal();
}

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;
  // 因为取数据的代码涉及到数据竞争 也就是说多个线程同时竞争 数组数据items 因此需要用锁保护起来
  lock.lockInterruptibly();
  try {
    // 当 count == 0 说明队列当中没有数据
    while (count == 0)
      notEmpty.await();
    // 当队列当中还有数据的时候可以将数据出队
    return 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; // 将对应的位置设置为 null 数据就可以被垃圾收集器回收了
  if (++takeIndex == items.length)
    takeIndex = 0;
  count--;
  // 迭代器也需要出队 如果不了
  if (itrs != null)
    itrs.elementDequeued();
  // 调用 signal 函数 将被 put 函数阻塞的线程唤醒 如果调用这个方法的时候没有线程阻塞
  // 那么调用这个方法相当于没有调用 如果有线程阻塞那么将会唤醒一个线程
  notFull.signal();
  return x;
}

public boolean offer(E e) {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  lock.lock();
  try {
    // 如果队列当中的数据个数和数组的长度相等 说明队列满了 直接返回 false 即可
    if (count == items.length)
      return false;
    else {
      enqueue(e);
      return true;
    }
  } finally {
    lock.unlock();
  }
}

public boolean add(E e) {
  if (offer(e))
    return true;
  else
    throw new IllegalStateException("Queue full");
}

public E poll() {
  final ReentrantLock lock = this.lock;
  lock.lock();
  try {
    return (count == 0) ? null : dequeue();
  } finally {
    lock.unlock();
  }
}