ArrayBlockingQueue

转 ：https://www.jb51.net/article/137750.htm

 

怎么使用数组来实现队列，我们需要四个变量：Object[] array来存储队列中元素，headIndex和tailIndex分别记录队列头和队列尾，count记录队列的个数。

1.因为数组的长度是固定，所以当count==array.length时，表示队列已经满了，当count==0的时候，表示队列是空的。

2.当添加元素的时候，将array[tailIndex] = e将tailIndex位置设置成新元素，之后将tailIndex++自增，然后将count++自增。但是有两点需要注意，在添加之前必须先判断队列是否已满，不然会出现覆盖已有元素。当tailIndex的值等于数组最后一个位置的时候，需要将tailIndex=0，循环利用数组

3.当删除元素的时候，将先记录下array[headIndex] 元素，之后将headIndex++自增，然后将count--自减。但是有两点需要注意要注意，在删除之前，必须先判断队列是否为空，不然可能会删除已删除的元素。

这里用了一个很巧妙的方式，我们知道当向队列中插入一个元素，那么就占用了数组的一个位置，当删除一个元素的时候，那么其实数组的这个位置就空闲出来了，表示这个位置又可以插入新元素了。

所以我们插入新元素前，必须检查队列是否已满，删除元素之前，必须检查队列是否为空。

二.ArrayBlockingQueue中重要成员变量

/** 储存队列的中元素 */
 final Object[] items;
 
 /** 队列头的位置 */
 int takeIndex;
 
 /** 队列尾的位置 */
 int putIndex;
 
 /** 当前队列拥有的元素个数 */
 int count;
 
 /** 用来保证多线程操作共享变量的安全问题 */
 final ReentrantLock lock;
 
 /** 当队列为空时，就会调用notEmpty的wait方法，让当前线程等待 */
 private final Condition notEmpty;
 
 /** 当队列为满时，就会调用notFull的wait方法，让当前线程等待 */
 private final Condition notFull;

就是多了lock、notEmpty、notFull变量来实现多线程安全和线程等待条件的，它们三个是怎么操作的呢？

 2.1 lock的作用

因为ArrayBlockingQueue是在多线程下操作的，所以修改items、takeIndex、putIndex和count这些成员变量时，必须要考虑多线程安全问题，所以这里使用lock独占锁，来保证并发操作的安全。

2.2 notEmpty与notFull的作用

因为阻塞队列必须实现，当队列为空或队列已满的时候，队列的读取或插入操作要等待。所以我们想到了并发框架下的Condition对象，使用它来控制。

在AQS中，我们介绍了这个类的作用：

1.await系列方法，会释放当前锁，并让当前线程等待。

2.signal与signalAll方法，会唤醒当前线程。其实它并不是唤醒当前线程，而是将在这个Condition条件上等待的线程，添加到lock锁上的等待线程池中，所以当锁被释放时，会唤醒lock锁上的等待线程池中一个线程。具体在AQS中有源码分析。

三. 添加元素方法

3.1 add(E e)与offer(E e)方法：

// 调用AbstractQueue父类中的方法。
public boolean add(E e) {
  // 通过调用offer来时实现
  if (offer(e))
    return true;
  else
    throw new IllegalStateException("Queue full");
}
 
//向队列末尾新添加元素。返回true表示添加成功，false表示添加失败，不会抛出异常
public boolean offer(E e) {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  // 使用lock来保证，多线程修改成员属性的安全
  lock.lock();
  try {
    // 队列已满，添加元素失败，返回false。
    if (count == items.length)
      return false;
    else {
      // 调用enqueue方法将元素插入队列中
      enqueue(e);
      return true;
    }
  } finally {
    lock.unlock();
  }
}

add方法是调用offer方法实现的。在offer方法中，必须先判断队列是否已满，如果已满就直接返回false，而不会阻塞当前线程。如果不满就调用enqueue方法将元素插入队列中。

注意：这里使用lock.lock()是保证同一时间只有一个线程修改成员变量，防止出现并发操作问题。虽然它也会阻塞当前线程，但是它并不是条件等待，只是因为锁被其他线程持有，而ArrayBlockingQueue中方法操作时间都不长，这里相当于不阻塞线程。

3.2 put方法

// 向队列末尾新添加元素，如果队列已满，当前线程就等待。响应中断异常
public void put(E e) throws InterruptedException {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  // 使用lock来保证，多线程修改成员属性的安全
  lock.lockInterruptibly();
  try {
    // 队列已满，则调用notFull.await()方法，让当前线程等待，直到队列不是满的
    while (count == items.length)
      notFull.await();
    // 调用enqueue方法将元素插入队列中
    enqueue(e);
  } finally {
    lock.unlock();
  }
}

与offer方法大体流程一样，只是当队列已满的时候，会调用notFull.await()方法，让当前线程阻塞等待，直到队列被别的线程移除了元素，队列不满的时候，会唤醒这个等待线程。

 3.3 offer(E e, long timeout, TimeUnit unit)方法

/**
   * 向队列末尾新添加元素，如果队列中没有可用空间，当前线程就等待，
   * 如果等待时间超过timeout了，那么返回false，表示添加失败
   */
  public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
 
    checkNotNull(e);
    // 计算一共最多等待的时间值nanos
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    // 使用lock来保证，多线程修改成员属性的安全
    lock.lockInterruptibly();
    try {
      // 队列已满
      while (count == items.length) {
        // nanos <= 0表示最大等待时间已到，那么不用再等待了，返回false，表示添加失败。
        if (nanos <= 0)
          return false;
        // 调用notFull.awaitNanos(nanos)方法,超时nanos时间会被自动唤醒，
        // 如果被提前唤醒，那么返回剩余的时间
        nanos = notFull.awaitNanos(nanos);
      }
      // 调用enqueue方法将元素插入队列中
      enqueue(e);
      return true;
    } finally {
      lock.unlock();
    }
  }

与put的方法大体流程一样，只不过是调用notFull.awaitNanos(nanos)方法，让当前线程等待，并设置等待时间。

四. 删除元素方法

4.1 remove()和poll()方法：

 // 调用AbstractQueue父类中的方法。
  public E remove() {
    // 通过调用poll来时实现
    E x = poll();
    if (x != null)
      return x;
    else
      throw new NoSuchElementException();
  }
 
// 删除队列第一个元素(即队列头)，并返回它。如果队列是空的，它不会抛出异常，而是会返回null。
  public E poll() {
    final ReentrantLock lock = this.lock;
    // 使用lock来保证，多线程修改成员属性的安全
    lock.lock();
    try {
      // 如果count == 0，列表为空，就返回null，否则调用dequeue方法，返回列表头元素
      return (count == 0) ? null : dequeue();
    } finally {
      lock.unlock();
    }
  }

 remove方法是调用poll()方法实现的。在 poll()方法中，如果列表为空，就返回null，否则调用dequeue方法，返回列表头元素。

4.2 take()方法

/**
 * 返回并移除队列第一个元素，如果队列是空的，就前线程就等待。响应中断异常
 */
public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
  // 使用lock来保证，多线程修改成员属性的安全
  lock.lockInterruptibly();
  try {
    // 如果队列为空，就调用notEmpty.await()方法，让当前线程等待。
    // 直到有别的线程向队列中插入元素，那么这个线程会被唤醒。
    while (count == 0)
      notEmpty.await();
    // 调用dequeue方法，返回列表头元素
    return dequeue();
  } finally {
    lock.unlock();
  }
}

take()方法当队列为空的时候，当前线程必须等待，直到有别的线程向队列中插入新元素，那么这个线程会被唤醒。

4.3 poll(long timeout, TimeUnit unit)方法

/**
 * 返回并移除队列第一个元素，如果队列是空的，就前线程就等待。
 * 如果等待时间超过timeout了，那么返回false，表示获取元素失败
 */
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
  // 计算一共最多等待的时间值nanos
  long nanos = unit.toNanos(timeout);
  final ReentrantLock lock = this.lock;
  // 使用lock来保证，多线程修改成员属性的安全
  lock.lockInterruptibly();
  try {
    // 队列为空
    while (count == 0) {
      // nanos <= 0表示最大等待时间已到，那么不用再等待了，返回null。
      if (nanos <= 0)
        return null;
      // 调用notEmpty.awaitNanos(nanos)方法让档期线程等待，并设置超时时间。
      nanos = notEmpty.awaitNanos(nanos);
    }
    // 调用dequeue方法，返回列表头元素
    return dequeue();
  } finally {
    lock.unlock();
  }
}

与take()方法流程一样，只不过调用awaitNanos(nanos)方法，让当前线程等待，并设置等待时间。

五. 查看元素的方法

5.1 element()与peek() 方法

// 调用AbstractQueue父类中的方法。
public E element() {
  E x = peek();
  if (x != null)
    return x;
  else
    throw new NoSuchElementException();
}
 
// 查看队列头元素
public E peek() {
  final ReentrantLock lock = this.lock;
  // 使用lock来保证，多线程修改成员属性的安全
  lock.lock();
  try {
    // 返回当前队列头的元素
    return itemAt(takeIndex); // null when queue is empty
  } finally {
    lock.unlock();
  }
}

六. 其他重要方法

6.1 enqueue和dequeue方法

// 将元素x插入队列尾
private void enqueue(E x) {
  // assert lock.getHoldCount() == 1;
  // assert items[putIndex] == null; //当前putIndex位置元素一定是null
  final Object[] items = this.items;
  items[putIndex] = x;
  // 如果putIndex等于items.length，那么将putIndex重新设置为0
  if (++putIndex == items.length)
    putIndex = 0;
  // 队列数量加一
  count++;
  // 因为插入一个元素，那么当前队列肯定不为空，那么唤醒在notEmpty条件下等待的一个线程
  notEmpty.signal();
}
 
// 删除队列头的元素，返回它
private E dequeue() {
  // assert lock.getHoldCount() == 1;
  // assert items[takeIndex] != null;
 
  final Object[] items = this.items;
  // 得到当前队列头的元素
  @SuppressWarnings("unchecked")
  E x = (E) items[takeIndex];
  // 将当前队列头位置设置为null
  items[takeIndex] = null;
  if (++takeIndex == items.length)
    takeIndex = 0;
  // 队列数量减一
  count--;
  if (itrs != null)
    itrs.elementDequeued();
  // 因为删除了一个元素，那么队列肯定不满了，那么唤醒在notFull条件下等待的一个线程
  notFull.signal();
  return x;
}

这两个方法分别代表，向队列中插入元素和从队列中删除元素。而且它们要唤醒等待的线程。当插入元素后，那么队列一定不为空，那么就要唤醒在notEmpty条件下等待的线程。当删除元素后，那么队列一定不满，那么就要唤醒在notFull条件下等待的线程。

6.2 remove(Object o)方法

// 删除队列中对象o元素，最多删除一条
public boolean remove(Object o) {
  if (o == null) return false;
  final Object[] items = this.items;
  // 使用lock来保证，多线程修改成员属性的安全
  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])) {
          // 删除i位置元素
          removeAt(i);
          // 返回true
          return true;
        }
        if (++i == items.length)
          i = 0;
        // 当i==putIndex表示遍历完所有元素
      } while (i != putIndex);
    }
    return false;
  } finally {
    lock.unlock();
  }
}

从队列中删除指定对象o，那么就要遍历队列，删除第一个与对象o相同的元素，如果队列中没有对象o元素，那么返回false删除失败。

这里有两点需要注意：

如何遍历队列，就是从队列头遍历到队列尾。就要靠takeIndex和putIndex两个变量了。

为什么Object[] items = this.items;这句代码没有放到同步锁lock代码块内。items是成员变量，那么多线程操作的时候，不会有并发问题么？

这个是因为items是个引用变量，不是基本数据类型，而且我们对队列的插入和删除操作，都是针对这一个items数组，没有改变数组的引用，所以在lock代码中，items会得到其他线程对它最新的修改。但是如果这里将int putIndex = this.putIndex;方法lock代码块外面，就会产生问题。

removeAt(final int removeIndex)方法

// 删除队列removeIndex位置的元素
void removeAt(final int removeIndex) {
  // assert lock.getHoldCount() == 1;
  // assert items[removeIndex] != null;
  // assert removeIndex >= 0 && removeIndex < items.length;
  final Object[] items = this.items;
  // 表示删除元素是列表头，就容易多了，与dequeue方法流程差不多
  if (removeIndex == takeIndex) {
    // 移除removeIndex位置元素
    items[takeIndex] = null;
    // 到了数组末尾，就要转到数组头位置
    if (++takeIndex == items.length)
      takeIndex = 0;
    // 队列数量减一
    count--;
    if (itrs != null)
      itrs.elementDequeued();
  } else {
    // an "interior" remove
 
    final int putIndex = this.putIndex;
    for (int i = removeIndex;;) {
      int next = i + 1;
      if (next == items.length)
        next = 0;
      // 还没有到队列尾，那么就将后一个位置元素覆盖前一个位置的元素
      if (next != putIndex) {
        items[i] = items[next];
        i = next;
      } else {
        // 将队列尾元素置位null
        items[i] = null;
        // 重新设置putIndex的值
        this.putIndex = i;
        break;
      }
    }
    // 队列数量减一
    count--;
    if (itrs != null)
      itrs.removedAt(removeIndex);
  }
  // 因为删除了一个元素，那么队列肯定不满了，那么唤醒在notFull条件下等待的一个线程
  notFull.signal();
}

在队列中删除指定位置的元素。需要注意的是删除之后的数组还能保持队列形式，分为两种情况：

1. 如果删除位置是队列头，那么简单，只需要将队列头的位置元素设置为null，将将队列头位置加一。
2. 如果删除位置不是队列头，那么麻烦了，这个时候，我们就要将从removeIndex位置后的元素全部左移一位，覆盖前一个元素。最后将原来队列尾的元素置位null。